feat: Add Spark SQL as a query engine

.gitignore

@@ -2,6 +2,7 @@
 ## SQRL
 ##############################
 plan-output/
+sqrl_iceberg_data
 
 ##############################
 ## Java

sqrl-planner/src/main/java/com/datasqrl/calcite/Dialect.java

@@ -21,5 +21,6 @@ public enum Dialect {
   FLINK,
   POSTGRES,
   SNOWFLAKE,
-  DUCKDB
+  DUCKDB,
+  SPARK_SQL
 }

sqrl-planner/src/main/java/com/datasqrl/calcite/convert/AbstractSqlConverters.java

@@ -62,14 +62,15 @@ public final Dialect getDialect() {
     return dialect;
   }
 
+  @Override
+  public final SqlDialect getCalciteSqlDialect() {
+    return calciteSqlDialect;
+  }
+
   protected SqlPrettyWriter createWriter() {
     var baseConfig = SqlPrettyWriter.config().withDialect(calciteSqlDialect);
     var config = SqrlConfigurations.SQL_TO_STRING.apply(baseConfig);
 
     return new DynamicParamSqlPrettyWriter(config);
   }
-
-  protected final SqlDialect getCalciteSqlDialect() {
-    return calciteSqlDialect;
-  }
 }

sqrl-planner/src/main/java/com/datasqrl/calcite/convert/FlinkSqlConverters.java

@@ -17,9 +17,11 @@
 
 import com.datasqrl.calcite.Dialect;
 import com.datasqrl.engine.stream.flink.sql.RelToFlinkSql;
+import com.datasqrl.engine.stream.flink.sql.calcite.FlinkDialect;
 import com.google.auto.service.AutoService;
 import java.util.Map;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.sql.SqlDialect;
 import org.apache.calcite.sql.SqlNode;
 
 @AutoService(SqlConverters.class)
@@ -36,6 +38,11 @@ public String convert(SqlNode sqlNode) {
     return RelToFlinkSql.convertToString(sqlNode);
   }
 
+  @Override
+  public SqlDialect getCalciteSqlDialect() {
+    return FlinkDialect.DEFAULT;
+  }
+
   @Override
   public Dialect getDialect() {
     return Dialect.FLINK;

sqrl-planner/src/main/java/com/datasqrl/calcite/convert/SparkSqlSqlConverters.java

@@ -0,0 +1,28 @@
+/*
+ * 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.calcite.convert;
+
+import com.datasqrl.calcite.Dialect;
+import com.datasqrl.calcite.dialect.ExtendedSparkSqlDialect;
+import com.google.auto.service.AutoService;
+
+@AutoService(SqlConverters.class)
+public class SparkSqlSqlConverters extends AbstractSqlConverters {
+
+  public SparkSqlSqlConverters() {
+    super(Dialect.SPARK_SQL, ExtendedSparkSqlDialect.DEFAULT, false);
+  }
+}

sqrl-planner/src/main/java/com/datasqrl/calcite/convert/SqlConverters.java

@@ -18,6 +18,7 @@
 import com.datasqrl.calcite.Dialect;
 import java.util.Map;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.sql.SqlDialect;
 import org.apache.calcite.sql.SqlNode;
 
 /** Provides conversions to SQL representations for a specific dialect. */
@@ -40,5 +41,12 @@ public interface SqlConverters {
    */
   String convert(SqlNode sqlNode);
 
+  /**
+   * Returns the Calcite dialect used for conversion and dialect-specific DDL generation.
+   *
+   * @return the configured Calcite SQL dialect
+   */
+  SqlDialect getCalciteSqlDialect();
+
   Dialect getDialect();
 }

sqrl-planner/src/main/java/com/datasqrl/calcite/dialect/ExtendedSparkSqlDialect.java

@@ -0,0 +1,93 @@
+/*
+ * 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.calcite.dialect;
+
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.sql.SqlAlienSystemTypeNameSpec;
+import org.apache.calcite.sql.SqlDataTypeSpec;
+import org.apache.calcite.sql.SqlDialect;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.dialect.SparkSqlDialect;
+import org.apache.calcite.sql.parser.SqlParserPos;
+
+public class ExtendedSparkSqlDialect extends SparkSqlDialect {
+
+  public static final Context DEFAULT_CONTEXT =
+      SqlDialect.EMPTY_CONTEXT
+          .withDatabaseProduct(DatabaseProduct.SPARK)
+          .withIdentifierQuoteString("`");
+
+  public static final SqlDialect DEFAULT = new ExtendedSparkSqlDialect(DEFAULT_CONTEXT);
+
+  public ExtendedSparkSqlDialect(Context context) {
+    super(context);
+  }
+
+  @Override
+  public SqlNode getCastSpec(RelDataType type) {
+    var castSpec = getSparkType(type);
+
+    return new SqlDataTypeSpec(
+        new SqlAlienSystemTypeNameSpec(castSpec, type.getSqlTypeName(), SqlParserPos.ZERO),
+        SqlParserPos.ZERO);
+  }
+
+  private String getSparkType(RelDataType type) {
+    String castSpec;
+    switch (type.getSqlTypeName()) {
+      case CHAR:
+      case VARCHAR:
+        castSpec = "STRING";
+        break;
+      case BINARY:
+      case VARBINARY:
+        castSpec = "BINARY";
+        break;
+      case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+        castSpec = "TIMESTAMP_LTZ";
+        break;
+      case TIMESTAMP:
+        castSpec = "TIMESTAMP_NTZ";
+        break;
+      case ARRAY:
+        castSpec = "ARRAY<" + getSparkType(type.getComponentType()) + ">";
+        break;
+      case MAP:
+        castSpec =
+            "MAP<"
+                + getSparkType(type.getKeyType())
+                + ", "
+                + getSparkType(type.getValueType())
+                + ">";
+        break;
+      case ROW:
+        castSpec =
+            "STRUCT<"
+                + type.getFieldList().stream()
+                    .map(
+                        field ->
+                            quoteIdentifier(field.getName()) + ": " + getSparkType(field.getType()))
+                    .reduce((left, right) -> left + ", " + right)
+                    .orElse("")
+                + ">";
+        break;
+      default:
+        return super.getCastSpec(type).toString();
+    }
+
+    return castSpec;
+  }
+}

sqrl-planner/src/main/java/com/datasqrl/config/QueryEngineConfigConverterImpl.java

@@ -36,14 +36,19 @@ public List<ObjectNode> convertConfigsToJson() {
 
     for (var engine : enabledQueryEngines) {
       var queryEngine = (QueryEngine) engine;
-      var engineConf = packageJson.getEngines().getEngineConfig(queryEngine.getName()).get();
+      var engineConf = packageJson.getEngines().getEngineConfig(queryEngine.getName());
+      if (engineConf.isEmpty()) {
+        continue;
+      }
 
-      if (engineConf instanceof EngineConfigImpl impl) {
+      if (engineConf.get() instanceof EngineConfigImpl impl) {
         var engineConfigMap = impl.sqrlConfig.toMap();
 
         var rootNode = JsonUtils.MAPPER.createObjectNode();
         var configNode = JsonUtils.MAPPER.valueToTree(engineConfigMap);
-        rootNode.set(queryEngine.serverConfigName(), configNode);
+        queryEngine
+            .serverConfigName()
+            .ifPresent(engineConfigName -> rootNode.set(engineConfigName, configNode));
 
         convertedConfigs.add(rootNode);
       }

sqrl-planner/src/main/java/com/datasqrl/engine/database/QueryEngine.java

@@ -18,6 +18,7 @@
 import com.datasqrl.engine.EnginePhysicalPlan;
 import com.datasqrl.engine.ExecutionEngine;
 import com.datasqrl.planner.dag.plan.MaterializationStagePlan;
+import java.util.Optional;
 
 /**
  * A {@link QueryEngine} executes queries against a {@link DatabaseEngine} that supports the query
@@ -28,5 +29,7 @@ public interface QueryEngine extends ExecutionEngine {
 
   EnginePhysicalPlan plan(MaterializationStagePlan stagePlan);
 
-  String serverConfigName();
+  default Optional<String> serverConfigName() {
+    return Optional.empty();
+  }
 }

sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java

@@ -57,7 +57,6 @@
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexShuttle;
-import org.apache.calcite.sql.SqlDataTypeSpec;
 import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlNode;
 import org.apache.calcite.sql.SqlNodeList;
@@ -83,6 +82,8 @@ protected AbstractJdbcStatementFactory(
         createTableDdlFactory);
   }
 
+  protected abstract SqlNode getSqlType(RelDataType type, Optional<DataTypeHint> hint);
+
   @Override
   public QueryResult createQuery(
       Query query, boolean withView, Map<String, JdbcEngineCreateTable> tableIdMap) {
@@ -120,7 +121,7 @@ public QueryResult createPassthroughQuery(Query query, boolean withView) {
     var viewName = query.function().getSimpleName();
     var rowType = query.relNode().getRowType();
     var viewSql =
-        new GenericCreateViewDdlFactory()
+        new GenericCreateViewDdlFactory(sqlConverters.getCalciteSqlDialect())
             .createView(viewName, rowType.getFieldNames(), passthroughSql);
     var view =
         new GenericJdbcStatement(
@@ -224,8 +225,6 @@ protected Field toField(RelDataTypeField field, PlannerHints hints, Documentatio
         documentation.getColumn(field.getName(), null));
   }
 
-  protected abstract SqlDataTypeSpec getSqlType(RelDataType type, Optional<DataTypeHint> hint);
-
   protected String createView(
       SqlIdentifier viewNameIdentifier, SqlNodeList columnList, SqlNode viewSqlNode) {
     var createView =
@@ -235,22 +234,6 @@ protected String createView(
     return sqlConverters.convert(createView);
   }
 
-  public static List<String> quoteIdentifier(List<String> columns) {
-    return columns.stream()
-        .map(AbstractJdbcStatementFactory::quoteIdentifier)
-        .collect(Collectors.toList());
-  }
-
-  public static String quoteIdentifier(String column) {
-    return "\"" + column + "\"";
-  }
-
-  public static List<String> quoteIdentifiers(List<String> values) {
-    return values.stream()
-        .map(AbstractJdbcStatementFactory::quoteIdentifier)
-        .collect(Collectors.toList());
-  }
-
   protected Set<DatabaseTypeExtension> extractTypeExtensions(
       Stream<RelNode> relNodes, List<DatabaseTypeExtension> extensions) {
     return relNodes

sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/DuckDBEngine.java

@@ -20,6 +20,7 @@
 import com.datasqrl.config.PackageJson;
 import com.datasqrl.server.jdbc.DatabaseType;
 import jakarta.inject.Inject;
+import java.util.Optional;
 import lombok.NonNull;
 
 public class DuckDBEngine extends AbstractJDBCQueryEngine {
@@ -33,8 +34,8 @@ public DuckDBEngine(@NonNull PackageJson json, ConnectorFactoryFactory connector
   }
 
   @Override
-  public String serverConfigName() {
-    return "duckDbConfig";
+  public Optional<String> serverConfigName() {
+    return Optional.of("duckDbConfig");
   }
 
   @Override
@@ -49,6 +50,6 @@ protected DatabaseType getDatabaseType() {
 
   @Override
   public JdbcStatementFactory getStatementFactory() {
-    return new DuckDbStatementFactory(engineConfig);
+    return new DuckDbStatementFactory();
   }
 }

sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/DuckDbStatementFactory.java

@@ -26,7 +26,6 @@
 import com.datasqrl.calcite.dialect.DuckDbSqlDialect;
 import com.datasqrl.calcite.type.TypeFactory;
 import com.datasqrl.config.JdbcDialect;
-import com.datasqrl.config.PackageJson.EngineConfig;
 import com.datasqrl.engine.database.relational.ddl.GenericCreateTableDdlFactory;
 import com.datasqrl.plan.global.IndexDefinition;
 import com.datasqrl.planner.dag.plan.MaterializationStagePlan.Query;
@@ -51,13 +50,11 @@
 
 public class DuckDbStatementFactory extends AbstractJdbcStatementFactory {
 
-  private final EngineConfig engineConfig;
-
-  public DuckDbStatementFactory(EngineConfig engineConfig) {
+  public DuckDbStatementFactory() {
     super(
         Dialect.DUCKDB,
-        new GenericCreateTableDdlFactory()); // Iceberg creates the tables, DuckDB only queries
-    this.engineConfig = engineConfig;
+        new GenericCreateTableDdlFactory(
+            DuckDbSqlDialect.DEFAULT)); // Iceberg creates the tables, DuckDB only queries
   }
 
   @Override

sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/IcebergEngine.java

@@ -58,7 +58,9 @@ public IcebergEngine(@NonNull PackageJson json, ConnectorFactoryFactory connecto
 
   @Override
   public boolean supportsQueryEngine(QueryEngine engine) {
-    return engine instanceof SnowflakeEngine || engine instanceof DuckDBEngine;
+    return engine instanceof SnowflakeEngine
+        || engine instanceof DuckDBEngine
+        || engine instanceof SparkSqlEngine;
   }
 
   @Override

sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java

@@ -124,7 +124,11 @@ public List<JdbcStatement> extractTypeExtensions(List<Query> queries) {
   public JdbcStatement addIndex(IndexDefinition index) {
     var ddl =
         new CreateIndexDDL(
-            index.getName(), index.getTableName(), index.getColumnNames(), index.getType());
+            index.getName(),
+            index.getTableName(),
+            index.getColumnNames(),
+            index.getType(),
+            ExtendedPostgresSqlDialect.DEFAULT);
     return new GenericJdbcStatement(ddl.getIndexName(), Type.INDEX, ddl.getSql());
   }
 

sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SnowflakeEngine.java

@@ -20,6 +20,7 @@
 import com.datasqrl.config.PackageJson;
 import com.datasqrl.server.jdbc.DatabaseType;
 import jakarta.inject.Inject;
+import java.util.Optional;
 import lombok.NonNull;
 
 public class SnowflakeEngine extends AbstractJDBCQueryEngine {
@@ -33,8 +34,8 @@ public SnowflakeEngine(@NonNull PackageJson json, ConnectorFactoryFactory connec
   }
 
   @Override
-  public String serverConfigName() {
-    return "snowflakeConfig";
+  public Optional<String> serverConfigName() {
+    return Optional.of("snowflakeConfig");
   }
 
   @Override