diff --git a/sqrl-planner/src/main/java/com/datasqrl/calcite/Dialect.java b/sqrl-planner/src/main/java/com/datasqrl/calcite/Dialect.java index 1a67dc389..df964e781 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/calcite/Dialect.java +++ b/sqrl-planner/src/main/java/com/datasqrl/calcite/Dialect.java @@ -21,5 +21,6 @@ public enum Dialect { FLINK, POSTGRES, SNOWFLAKE, - DUCKDB + DUCKDB, + SPARK_SQL } diff --git a/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/AbstractSqlConverters.java b/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/AbstractSqlConverters.java index dfe40322c..8b6c48641 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/AbstractSqlConverters.java +++ b/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; - } } diff --git a/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/FlinkSqlConverters.java b/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/FlinkSqlConverters.java index ffc750d19..223f0a920 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/FlinkSqlConverters.java +++ b/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; diff --git a/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/SparkSqlSqlConverters.java b/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/SparkSqlSqlConverters.java new file mode 100644 index 000000000..93334b019 --- /dev/null +++ b/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); + } +} diff --git a/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/SqlConverters.java b/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/SqlConverters.java index 4ce3f60c5..bfb301be4 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/calcite/convert/SqlConverters.java +++ b/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(); } diff --git a/sqrl-planner/src/main/java/com/datasqrl/calcite/dialect/ExtendedSparkSqlDialect.java b/sqrl-planner/src/main/java/com/datasqrl/calcite/dialect/ExtendedSparkSqlDialect.java new file mode 100644 index 000000000..b1977e767 --- /dev/null +++ b/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; + } +} diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java index 1ab3df510..4dadbfa19 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/AbstractJdbcStatementFactory.java @@ -58,7 +58,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; @@ -84,6 +83,8 @@ protected AbstractJdbcStatementFactory( createTableDdlFactory); } + protected abstract SqlNode getSqlType(RelDataType type, Optional hint); + @Override public QueryResult createQuery( Query query, boolean withView, Map tableIdMap) { @@ -121,7 +122,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( @@ -232,8 +233,6 @@ protected Field toField(RelDataTypeField field, PlannerHints hints, Documentatio documentation.getColumn(field.getName(), null)); } - protected abstract SqlDataTypeSpec getSqlType(RelDataType type, Optional hint); - protected String createView( SqlIdentifier viewNameIdentifier, SqlNodeList columnList, SqlNode viewSqlNode) { var createView = @@ -243,22 +242,6 @@ protected String createView( return sqlConverters.convert(createView); } - public static List quoteIdentifier(List columns) { - return columns.stream() - .map(AbstractJdbcStatementFactory::quoteIdentifier) - .collect(Collectors.toList()); - } - - public static String quoteIdentifier(String column) { - return "\"" + column + "\""; - } - - public static List quoteIdentifiers(List values) { - return values.stream() - .map(AbstractJdbcStatementFactory::quoteIdentifier) - .collect(Collectors.toList()); - } - protected Set extractTypeExtensions( Stream relNodes, List extensions) { return relNodes diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/DuckDbStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/DuckDbStatementFactory.java index 099986097..2ea259cc2 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/DuckDbStatementFactory.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/DuckDbStatementFactory.java @@ -52,7 +52,8 @@ public class DuckDbStatementFactory extends AbstractJdbcStatementFactory { public DuckDbStatementFactory() { super( Dialect.DUCKDB, - new GenericCreateTableDdlFactory()); // Iceberg creates the tables, DuckDB only queries + new GenericCreateTableDdlFactory( + DuckDbSqlDialect.DEFAULT)); // Iceberg creates the tables, DuckDB only queries } @Override diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/IcebergEngine.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/IcebergEngine.java index 4770d17de..aa41f9824 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/IcebergEngine.java +++ b/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 diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java index 598c82dad..620ba7cf1 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/PostgresStatementFactory.java @@ -141,7 +141,11 @@ public List extractTypeExtensions(List 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()); } diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SnowflakeStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SnowflakeStatementFactory.java index a02a23e69..792a58397 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SnowflakeStatementFactory.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SnowflakeStatementFactory.java @@ -17,6 +17,7 @@ import com.datasqrl.calcite.Dialect; import com.datasqrl.calcite.dialect.ExtendedPostgresSqlDialect; +import com.datasqrl.calcite.dialect.ExtendedSnowflakeSqlDialect; import com.datasqrl.calcite.dialect.snowflake.SqlCreateIcebergTableFromObjectStorage; import com.datasqrl.config.PackageJson.EngineConfig; import com.datasqrl.deployment.model.JdbcStatementModel.Type; @@ -35,7 +36,7 @@ public class SnowflakeStatementFactory extends AbstractJdbcStatementFactory { private final EngineConfig engineConfig; public SnowflakeStatementFactory(EngineConfig engineConfig) { - super(Dialect.SNOWFLAKE, new GenericCreateTableDdlFactory()); + super(Dialect.SNOWFLAKE, new GenericCreateTableDdlFactory(ExtendedSnowflakeSqlDialect.DEFAULT)); this.engineConfig = engineConfig; } diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlEngine.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlEngine.java new file mode 100644 index 000000000..03e450d95 --- /dev/null +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlEngine.java @@ -0,0 +1,43 @@ +/* + * 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.engine.database.relational; + +import com.datasqrl.config.ConnectorFactoryFactory; +import com.datasqrl.config.JdbcDialect; +import com.datasqrl.config.PackageJson; +import jakarta.inject.Inject; +import lombok.NonNull; + +public class SparkSqlEngine extends AbstractJDBCShallowQueryEngine { + + @Inject + public SparkSqlEngine(@NonNull PackageJson json, ConnectorFactoryFactory connectorFactory) { + super( + SparkSqlEngineFactory.ENGINE_NAME, + json.getEngines().getEngineConfigOrEmpty(SparkSqlEngineFactory.ENGINE_NAME), + connectorFactory); + } + + @Override + protected JdbcDialect getDialect() { + return JdbcDialect.Iceberg; + } + + @Override + public JdbcStatementFactory getStatementFactory() { + return new SparkSqlStatementFactory(); + } +} diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlEngineFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlEngineFactory.java new file mode 100644 index 000000000..562821226 --- /dev/null +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlEngineFactory.java @@ -0,0 +1,36 @@ +/* + * 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.engine.database.relational; + +import com.datasqrl.config.EngineFactory; +import com.datasqrl.engine.database.DatabaseEngineFactory; +import com.google.auto.service.AutoService; + +@AutoService(EngineFactory.class) +public class SparkSqlEngineFactory implements DatabaseEngineFactory { + + public static final String ENGINE_NAME = "sparksql"; + + @Override + public String getEngineName() { + return ENGINE_NAME; + } + + @Override + public Class getFactoryClass() { + return SparkSqlEngine.class; + } +} diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlStatementFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlStatementFactory.java new file mode 100644 index 000000000..4834905c1 --- /dev/null +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/SparkSqlStatementFactory.java @@ -0,0 +1,42 @@ +/* + * 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.engine.database.relational; + +import com.datasqrl.calcite.Dialect; +import com.datasqrl.calcite.dialect.ExtendedSparkSqlDialect; +import com.datasqrl.engine.database.relational.ddl.GenericCreateTableDdlFactory; +import com.datasqrl.plan.global.IndexDefinition; +import com.datasqrl.planner.hint.DataTypeHint; +import java.util.Optional; +import org.apache.calcite.rel.type.RelDataType; +import org.apache.calcite.sql.SqlNode; + +public class SparkSqlStatementFactory extends AbstractJdbcStatementFactory { + + public SparkSqlStatementFactory() { + super(Dialect.SPARK_SQL, new GenericCreateTableDdlFactory(ExtendedSparkSqlDialect.DEFAULT)); + } + + @Override + protected SqlNode getSqlType(RelDataType type, Optional hint) { + return ExtendedSparkSqlDialect.DEFAULT.getCastSpec(type); + } + + @Override + public JdbcStatement addIndex(IndexDefinition indexDefinition) { + throw new UnsupportedOperationException("Spark SQL does not support indexes"); + } +} diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/CreateIndexDDL.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/CreateIndexDDL.java index a23ebc60d..adf39e85b 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/CreateIndexDDL.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/CreateIndexDDL.java @@ -15,14 +15,13 @@ */ package com.datasqrl.engine.database.relational.ddl; -import static com.datasqrl.engine.database.relational.AbstractJdbcStatementFactory.quoteIdentifier; - import com.datasqrl.plan.global.IndexType; import com.datasqrl.sql.SqlDDLStatement; import com.google.common.base.Preconditions; import java.util.List; import java.util.stream.Collectors; import lombok.Value; +import org.apache.calcite.sql.SqlDialect; @Value public class CreateIndexDDL implements SqlDDLStatement { @@ -31,6 +30,20 @@ public class CreateIndexDDL implements SqlDDLStatement { String tableName; List columns; IndexType type; + DdlIdentifierQuoter identifierQuoter; + + public CreateIndexDDL( + String indexName, + String tableName, + List columns, + IndexType type, + SqlDialect dialect) { + this.indexName = indexName; + this.tableName = tableName; + this.columns = columns; + this.type = type; + this.identifierQuoter = new DdlIdentifierQuoter(dialect); + } @Override public String getSql() { @@ -40,8 +53,8 @@ public String getSql() { columnExpression = "to_tsvector('english', %s )" .formatted( - quoteIdentifier(columns).stream() - .map(col -> "coalesce(%s, '')".formatted(col)) + identifierQuoter.quoteAll(columns).stream() + .map("coalesce(%s, '')"::formatted) .collect(Collectors.joining(" || ' ' || "))); indexType = "GIN"; break; @@ -55,18 +68,22 @@ public String getSql() { case VECTOR_EUCLID -> "vector_l2_ops"; default -> throw new UnsupportedOperationException(type.toString()); }; - columnExpression = quoteIdentifier(columns.get(0)) + " " + indexModifier; + columnExpression = identifierQuoter.quote(columns.get(0)) + " " + indexModifier; indexType = "HNSW"; break; default: - columnExpression = String.join(",", quoteIdentifier(columns)); + columnExpression = String.join(",", identifierQuoter.quoteAll(columns)); indexType = type.name().toLowerCase(); } var createTable = "CREATE INDEX IF NOT EXISTS %s ON %s USING %s (%s)"; var sql = createTable.formatted( - quoteIdentifier(indexName), quoteIdentifier(tableName), indexType, columnExpression); + identifierQuoter.quote(indexName), + identifierQuoter.quote(tableName), + indexType, + columnExpression); + return sql; } } diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/DdlIdentifierQuoter.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/DdlIdentifierQuoter.java new file mode 100644 index 000000000..538b9ee93 --- /dev/null +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/DdlIdentifierQuoter.java @@ -0,0 +1,47 @@ +/* + * 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.engine.database.relational.ddl; + +import jakarta.annotation.Nullable; +import java.util.List; +import lombok.RequiredArgsConstructor; +import org.apache.calcite.sql.SqlDialect; + +/** + * Quotes DDL identifiers using the configured Calcite SQL dialect, or double quotes when no dialect + * is configured. + */ +@RequiredArgsConstructor +public final class DdlIdentifierQuoter { + + @Nullable private final SqlDialect dialect; + + public DdlIdentifierQuoter() { + this(null); + } + + public String quote(String identifier) { + if (dialect != null) { + return dialect.quoteIdentifier(identifier); + } + + return "\"" + identifier + "\""; + } + + public List quoteAll(List identifiers) { + return identifiers.stream().map(this::quote).toList(); + } +} diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateTableDdlFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateTableDdlFactory.java index 03cbc8902..1b5d0891e 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateTableDdlFactory.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateTableDdlFactory.java @@ -15,28 +15,34 @@ */ package com.datasqrl.engine.database.relational.ddl; -import static com.datasqrl.engine.database.relational.AbstractJdbcStatementFactory.quoteIdentifier; -import static com.datasqrl.engine.database.relational.AbstractJdbcStatementFactory.quoteIdentifiers; - import com.datasqrl.deployment.model.JdbcStatementModel.Field; import com.datasqrl.engine.database.relational.CreateTableJdbcStatement; import com.datasqrl.engine.database.relational.CreateTableJdbcStatement.CreateTableDdlFactory; import java.util.List; import java.util.StringJoiner; +import org.apache.calcite.sql.SqlDialect; public class GenericCreateTableDdlFactory implements CreateTableDdlFactory { private static final String CREATE_TABLE_TEMPLATE = "CREATE TABLE IF NOT EXISTS %s (%s)"; + private final DdlIdentifierQuoter identifierQuoter; + + public GenericCreateTableDdlFactory(SqlDialect dialect) { + this.identifierQuoter = new DdlIdentifierQuoter(dialect); + } + + protected GenericCreateTableDdlFactory() { + this.identifierQuoter = new DdlIdentifierQuoter(); + } + @Override public String createTableDdl(CreateTableJdbcStatement stmt) { validatePartitionType(stmt.getPartitionType()); var tableElements = new StringJoiner(", "); // Add field definitions - stmt.getFields().stream() - .map(GenericCreateTableDdlFactory::fieldToSql) - .forEach(tableElements::add); + stmt.getFields().stream().map(this::fieldToSql).forEach(tableElements::add); // Add primary key constraint if present if (!stmt.getPrimaryKey().isEmpty()) { @@ -47,13 +53,21 @@ public String createTableDdl(CreateTableJdbcStatement stmt) { quoteIdentifier(stmt.getName()), tableElements.toString()); } - protected static String listToSql(List columns) { + protected String listToSql(List columns) { return String.join(",", quoteIdentifiers(columns)); } - protected static String fieldToSql(Field field) { + protected String fieldToSql(Field field) { var notNull = field.nullable() ? "" : "NOT NULL"; - return "\"%s\" %s %s".formatted(field.name(), field.type(), notNull).trim(); + return "%s %s %s".formatted(quoteIdentifier(field.name()), field.type(), notNull).trim(); + } + + protected String quoteIdentifier(String identifier) { + return identifierQuoter.quote(identifier); + } + + protected List quoteIdentifiers(List identifiers) { + return identifierQuoter.quoteAll(identifiers); } } diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateViewDdlFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateViewDdlFactory.java index 04e32923c..a27fdcdad 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateViewDdlFactory.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/GenericCreateViewDdlFactory.java @@ -15,21 +15,24 @@ */ package com.datasqrl.engine.database.relational.ddl; -import static com.datasqrl.engine.database.relational.AbstractJdbcStatementFactory.quoteIdentifier; - -import com.datasqrl.engine.database.relational.AbstractJdbcStatementFactory; import java.util.List; import java.util.stream.Collectors; +import lombok.RequiredArgsConstructor; +import org.apache.calcite.sql.SqlDialect; +@RequiredArgsConstructor public class GenericCreateViewDdlFactory { + private final DdlIdentifierQuoter identifierQuoter; + + public GenericCreateViewDdlFactory(SqlDialect dialect) { + this(new DdlIdentifierQuoter(dialect)); + } + public String createView(String viewName, List columns, String select) { - var colStr = - columns.stream() - .map(AbstractJdbcStatementFactory::quoteIdentifier) - .collect(Collectors.joining(", ")); + var colStr = columns.stream().map(identifierQuoter::quote).collect(Collectors.joining(", ")); return "CREATE OR REPLACE VIEW %s (%s) AS %s" - .formatted(quoteIdentifier(viewName), colStr, select); + .formatted(identifierQuoter.quote(viewName), colStr, select); } } diff --git a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactory.java b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactory.java index e49cf24ca..7e279da81 100644 --- a/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactory.java +++ b/sqrl-planner/src/main/java/com/datasqrl/engine/database/relational/ddl/PostgresCreateTableDdlFactory.java @@ -15,17 +15,14 @@ */ package com.datasqrl.engine.database.relational.ddl; -import static com.datasqrl.engine.database.relational.AbstractJdbcStatementFactory.quoteIdentifier; - +import com.datasqrl.calcite.dialect.ExtendedPostgresSqlDialect; import com.datasqrl.deployment.model.JdbcStatementModel.PartitionType; import com.datasqrl.engine.database.relational.CreateTableJdbcStatement; import java.time.Duration; import java.util.EnumSet; import java.util.Optional; import java.util.Set; -import lombok.AllArgsConstructor; -@AllArgsConstructor public class PostgresCreateTableDdlFactory extends GenericCreateTableDdlFactory { private static final Set SUPPORTED_PARTITIONS = @@ -35,6 +32,11 @@ public class PostgresCreateTableDdlFactory extends GenericCreateTableDdlFactory private final boolean addDefaultPartition; + public PostgresCreateTableDdlFactory(boolean addDefaultPartition) { + super(ExtendedPostgresSqlDialect.DEFAULT); + this.addDefaultPartition = addDefaultPartition; + } + @Override public String createTableDdl(CreateTableJdbcStatement stmt) { var ddl = super.createTableDdl(stmt); diff --git a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/FullUseCaseIT.java b/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/FullUseCaseIT.java index aa43370e1..d0f02ed6b 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/FullUseCaseIT.java +++ b/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/FullUseCaseIT.java @@ -65,7 +65,8 @@ void specificUseCase(UseCaseParam param, TestContainerHook hook) { /** Ad-hoc debugging entry point. Change the path below to run a single use case manually. */ static Stream specificUseCaseProvider() { - return Stream.of(new UseCaseParam(USE_CASES.resolve("jwt-authorized").resolve("package.json"))); + return Stream.of( + new UseCaseParam(USE_CASES.resolve("complex-mutation").resolve("package.json"))); } @ParameterizedTest diff --git a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java b/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java index 14901bb02..f22d6e9aa 100644 --- a/sqrl-testing/sqrl-testing-integration/src/test/java/com/datasqrl/UseCaseCompileTest.java +++ b/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("jwt-authorized").resolve("package.json"); + var pkg = USECASE_DIR.resolve("dialects/spark-sql-compile").resolve("package.json"); UseCaseTestHelper.testUseCase( snapshotExtension, getClass(), diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/engine-validation/package-sparksql-no-server.json b/sqrl-testing/sqrl-testing-integration/src/test/resources/engine-validation/package-sparksql-no-server.json new file mode 100644 index 000000000..c18815bfc --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/engine-validation/package-sparksql-no-server.json @@ -0,0 +1,14 @@ +{ + "version": "1", + "enabled-engines": ["flink", "iceberg", "sparksql"], + "script": { + "main": "script.sqrl" + }, + "engines": { + "snowflake": { + "catalog-name": "MyCatalog", + "external-volume": "MyNewVolume", + "url": "${SNOWFLAKE_JDBC_URL}" + } + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/EngineValidationTest/package-sparksql-no-server.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/EngineValidationTest/package-sparksql-no-server.txt new file mode 100644 index 000000000..4856287ef --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/EngineValidationTest/package-sparksql-no-server.txt @@ -0,0 +1,97 @@ +>>>iceberg-schema.sql +CREATE TABLE IF NOT EXISTS "Numbers" ("id" INTEGER NOT NULL, PRIMARY KEY ("id")); +CREATE TABLE IF NOT EXISTS "Numbers2" ("id" INTEGER NOT NULL, PRIMARY KEY ("id")) +>>>iceberg-sparksql-schema.sql +CREATE TABLE IF NOT EXISTS `Numbers` (`id` INTEGER NOT NULL, PRIMARY KEY (`id`)); +CREATE TABLE IF NOT EXISTS `Numbers2` (`id` INTEGER NOT NULL, PRIMARY KEY (`id`)) +>>>iceberg.json +{ + "plans" : { + "" : { + "statements" : [ + { + "name" : "Numbers", + "type" : "TABLE", + "sql" : "CREATE TABLE IF NOT EXISTS \"Numbers\" (\"id\" INTEGER NOT NULL, PRIMARY KEY (\"id\"))", + "fields" : [ + { + "name" : "id", + "type" : "INTEGER", + "nullable" : false + } + ], + "primaryKey" : [ + "id" + ], + "partitionKey" : [ ], + "partitionType" : "NONE", + "numPartitions" : 0, + "ttl" : 0.0 + }, + { + "name" : "Numbers2", + "type" : "TABLE", + "sql" : "CREATE TABLE IF NOT EXISTS \"Numbers2\" (\"id\" INTEGER NOT NULL, PRIMARY KEY (\"id\"))", + "fields" : [ + { + "name" : "id", + "type" : "INTEGER", + "nullable" : false + } + ], + "primaryKey" : [ + "id" + ], + "partitionKey" : [ ], + "partitionType" : "NONE", + "numPartitions" : 0, + "ttl" : 0.0 + } + ], + "standaloneExtensionStatements" : [ ] + }, + "sparksql" : { + "statements" : [ + { + "name" : "Numbers", + "type" : "TABLE", + "sql" : "CREATE TABLE IF NOT EXISTS `Numbers` (`id` INTEGER NOT NULL, PRIMARY KEY (`id`))", + "fields" : [ + { + "name" : "id", + "type" : "INTEGER", + "nullable" : false + } + ], + "primaryKey" : [ + "id" + ], + "partitionKey" : [ ], + "partitionType" : "NONE", + "numPartitions" : 0, + "ttl" : 0.0 + }, + { + "name" : "Numbers2", + "type" : "TABLE", + "sql" : "CREATE TABLE IF NOT EXISTS `Numbers2` (`id` INTEGER NOT NULL, PRIMARY KEY (`id`))", + "fields" : [ + { + "name" : "id", + "type" : "INTEGER", + "nullable" : false + } + ], + "primaryKey" : [ + "id" + ], + "partitionKey" : [ ], + "partitionType" : "NONE", + "numPartitions" : 0, + "ttl" : 0.0 + } + ], + "standaloneExtensionStatements" : [ ] + } + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/dialects-spark-sql-compile-package.txt b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/dialects-spark-sql-compile-package.txt new file mode 100644 index 000000000..af77de400 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/dialects-spark-sql-compile-package.txt @@ -0,0 +1,205 @@ +>>>pipeline_explain.txt +=== SparkAggregates +ID: default_catalog.default_database.SparkAggregates +Type: state +Stage: flink +Primary key: boolean_value +Timestamp: - +Row count: ~1e7 +--- +Schema: + - boolean_value: BOOLEAN + - record_count: BIGINT NOT NULL + - decimal_sum: DECIMAL(38, 2) + - double_average: DOUBLE + - latest_event: TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) +Inputs: + - default_catalog.default_database.SparkTypeSource +Annotations: + - stream-root: SparkTypeSource + +=== SparkTypeMappings +ID: default_catalog.default_database.SparkTypeMappings +Type: stream +Stage: flink +Primary key: id +Timestamp: - +Row count: ~1e8 +--- +Schema: + - id: BIGINT NOT NULL + - tiny_value: TINYINT + - small_value: SMALLINT + - int_value: INTEGER + - float_value: FLOAT + - double_value: DOUBLE + - decimal_value: DECIMAL(12, 2) + - boolean_value: BOOLEAN + - string_value: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" + - varchar_value: VARCHAR(32) CHARACTER SET "UTF-16LE" + - char_value: CHAR(8) CHARACTER SET "UTF-16LE" + - binary_value: BINARY(16) + - varbinary_value: VARBINARY(32) + - date_value: DATE + - timestamp_value: TIMESTAMP(3) + - timestamp_ltz_value: TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) + - string_array: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" ARRAY + - string_to_int: (VARCHAR(2147483647) CHARACTER SET "UTF-16LE", INTEGER) MAP + - attributes: RecordType:peek_no_expand(VARCHAR(2147483647) CHARACTER SET "UTF-16LE" source, INTEGER priority) +Inputs: + - default_catalog.default_database.SparkTypeSource +Annotations: + - stream-root: SparkTypeSource + +=== SparkTypeSource +ID: default_catalog.default_database.SparkTypeSource +Type: stream +Stage: flink +Primary key: id +Timestamp: - +Row count: ~1e8 +--- +Schema: + - id: BIGINT NOT NULL + - tiny_value: TINYINT + - small_value: SMALLINT + - int_value: INTEGER + - float_value: FLOAT + - double_value: DOUBLE + - decimal_value: DECIMAL(12, 2) + - boolean_value: BOOLEAN + - string_value: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" + - varchar_value: VARCHAR(32) CHARACTER SET "UTF-16LE" + - char_value: CHAR(8) CHARACTER SET "UTF-16LE" + - binary_value: BINARY(16) + - varbinary_value: VARBINARY(32) + - date_value: DATE + - timestamp_value: TIMESTAMP(3) + - timestamp_ltz_value: TIMESTAMP_WITH_LOCAL_TIME_ZONE(3) + - string_array: VARCHAR(2147483647) CHARACTER SET "UTF-16LE" ARRAY + - string_to_int: (VARCHAR(2147483647) CHARACTER SET "UTF-16LE", INTEGER) MAP + - attributes: RecordType:peek_no_expand(VARCHAR(2147483647) CHARACTER SET "UTF-16LE" source, INTEGER priority) +Inputs: + - default_catalog.default_database.SparkTypeSource__base +Annotations: + - features: DENORMALIZE (feature) + - stream-root: SparkTypeSource + +>>>flink-sql-no-functions.sql +CREATE TABLE `SparkTypeSource` ( + `id` BIGINT NOT NULL, + `tiny_value` TINYINT, + `small_value` SMALLINT, + `int_value` INTEGER, + `float_value` FLOAT, + `double_value` DOUBLE, + `decimal_value` DECIMAL(12, 2), + `boolean_value` BOOLEAN, + `string_value` STRING, + `varchar_value` VARCHAR(32), + `char_value` CHAR(8), + `binary_value` BINARY(16), + `varbinary_value` VARBINARY(32), + `date_value` DATE, + `timestamp_value` TIMESTAMP(3), + `timestamp_ltz_value` TIMESTAMP_LTZ(3), + `string_array` ARRAY< STRING >, + `string_to_int` MAP< STRING, INTEGER >, + `attributes` ROW< `source` STRING, `priority` INTEGER >, + PRIMARY KEY (`id`) NOT ENFORCED +) +WITH ( + 'catalog-name' = 'default_catalog', + 'catalog-table' = 'SparkTypeSource', + 'catalog-type' = 'hadoop', + 'commit.retry.max-wait-ms' = '5000', + 'commit.retry.min-wait-ms' = '100', + 'commit.retry.num-retries' = '20', + 'connector' = 'iceberg', + 'format-version' = '2', + 'warehouse' = 'sqrl_iceberg_data', + 'write.distribution-mode' = 'hash', + 'write.upsert.enabled' = 'true' +); +CREATE VIEW `SparkTypeMappings` +AS +SELECT `id`, `tiny_value`, `small_value`, `int_value`, `float_value`, `double_value`, `decimal_value`, `boolean_value`, `string_value`, `varchar_value`, `char_value`, `binary_value`, `varbinary_value`, `date_value`, `timestamp_value`, `timestamp_ltz_value`, `string_array`, `string_to_int`, `attributes` +FROM `SparkTypeSource`; +CREATE VIEW `SparkAggregates` +AS +SELECT `boolean_value`, COUNT(*) AS `record_count`, SUM(`decimal_value`) AS `decimal_sum`, AVG(`double_value`) AS `double_average`, MAX(`timestamp_ltz_value`) AS `latest_event` +FROM `SparkTypeSource` +GROUP BY `boolean_value`; +CREATE TABLE `SparkAggregates_1` ( + `boolean_value` BOOLEAN, + `record_count` BIGINT NOT NULL, + `decimal_sum` DECIMAL(38, 2), + `double_average` DOUBLE, + `latest_event` TIMESTAMP(3) WITH LOCAL TIME ZONE, + PRIMARY KEY (`boolean_value`) NOT ENFORCED +) +WITH ( + 'catalog-name' = 'default_catalog', + 'catalog-table' = 'SparkAggregates', + 'catalog-type' = 'hadoop', + 'commit.retry.max-wait-ms' = '5000', + 'commit.retry.min-wait-ms' = '100', + 'commit.retry.num-retries' = '20', + 'connector' = 'iceberg', + 'format-version' = '2', + 'warehouse' = 'sqrl_iceberg_data', + 'write.distribution-mode' = 'hash', + 'write.upsert.enabled' = 'true' +); +CREATE TABLE `SparkTypeMappings_2` ( + `id` BIGINT NOT NULL, + `tiny_value` TINYINT, + `small_value` SMALLINT, + `int_value` INTEGER, + `float_value` FLOAT, + `double_value` DOUBLE, + `decimal_value` DECIMAL(12, 2), + `boolean_value` BOOLEAN, + `string_value` VARCHAR(2147483647) CHARACTER SET `UTF-16LE`, + `varchar_value` VARCHAR(32) CHARACTER SET `UTF-16LE`, + `char_value` CHAR(8) CHARACTER SET `UTF-16LE`, + `binary_value` BINARY(16), + `varbinary_value` VARBINARY(32), + `date_value` DATE, + `timestamp_value` TIMESTAMP(3), + `timestamp_ltz_value` TIMESTAMP(3) WITH LOCAL TIME ZONE, + `string_array` VARCHAR(2147483647) CHARACTER SET `UTF-16LE` ARRAY, + `string_to_int` MAP< VARCHAR(2147483647) CHARACTER SET `UTF-16LE`, INTEGER >, + `attributes` ROW(`source` VARCHAR(2147483647) CHARACTER SET `UTF-16LE`, `priority` INTEGER), + PRIMARY KEY (`id`) NOT ENFORCED +) +WITH ( + 'catalog-name' = 'default_catalog', + 'catalog-table' = 'SparkTypeMappings', + 'catalog-type' = 'hadoop', + 'commit.retry.max-wait-ms' = '5000', + 'commit.retry.min-wait-ms' = '100', + 'commit.retry.num-retries' = '20', + 'connector' = 'iceberg', + 'format-version' = '2', + 'warehouse' = 'sqrl_iceberg_data', + 'write.distribution-mode' = 'hash' +); +EXECUTE STATEMENT SET BEGIN +INSERT INTO `default_catalog`.`default_database`.`SparkAggregates_1` +SELECT * +FROM `default_catalog`.`default_database`.`SparkAggregates` +; +INSERT INTO `default_catalog`.`default_database`.`SparkTypeMappings_2` +SELECT * +FROM `default_catalog`.`default_database`.`SparkTypeMappings` +; +END +>>>iceberg-schema.sql +CREATE TABLE IF NOT EXISTS "SparkAggregates" ("boolean_value" BOOLEAN, "record_count" BIGINT NOT NULL, "decimal_sum" NUMERIC, "double_average" DOUBLE PRECISION, "latest_event" TIMESTAMP WITH TIME ZONE, PRIMARY KEY ("boolean_value")); +CREATE TABLE IF NOT EXISTS "SparkTypeMappings" ("id" BIGINT NOT NULL, "tiny_value" SMALLINT, "small_value" SMALLINT, "int_value" INTEGER, "float_value" FLOAT, "double_value" DOUBLE PRECISION, "decimal_value" NUMERIC, "boolean_value" BOOLEAN, "string_value" TEXT, "varchar_value" TEXT, "char_value" TEXT, "binary_value" BYTEA, "varbinary_value" BYTEA, "date_value" DATE, "timestamp_value" TIMESTAMP WITHOUT TIME ZONE, "timestamp_ltz_value" TIMESTAMP WITH TIME ZONE, "string_array" TEXT ARRAY, "string_to_int" JSONB, "attributes" JSONB, PRIMARY KEY ("id")); +CREATE TABLE IF NOT EXISTS "SparkTypeSource" ("id" BIGINT NOT NULL, "tiny_value" SMALLINT, "small_value" SMALLINT, "int_value" INTEGER, "float_value" FLOAT, "double_value" DOUBLE PRECISION, "decimal_value" NUMERIC, "boolean_value" BOOLEAN, "string_value" TEXT, "varchar_value" TEXT, "char_value" TEXT, "binary_value" BYTEA, "varbinary_value" BYTEA, "date_value" DATE, "timestamp_value" TIMESTAMP WITHOUT TIME ZONE, "timestamp_ltz_value" TIMESTAMP WITH TIME ZONE, "string_array" TEXT ARRAY, "string_to_int" JSONB, "attributes" JSONB, PRIMARY KEY ("id")) +>>>iceberg-sparksql-schema.sql +CREATE TABLE IF NOT EXISTS `SparkAggregates` (`boolean_value` BOOLEAN, `record_count` BIGINT NOT NULL, `decimal_sum` DECIMAL(19, 2), `double_average` DOUBLE, `latest_event` TIMESTAMP_LTZ, PRIMARY KEY (`boolean_value`)); +CREATE TABLE IF NOT EXISTS `SparkTypeMappings` (`id` BIGINT NOT NULL, `tiny_value` TINYINT, `small_value` SMALLINT, `int_value` INTEGER, `float_value` FLOAT, `double_value` DOUBLE, `decimal_value` DECIMAL(12, 2), `boolean_value` BOOLEAN, `string_value` STRING, `varchar_value` STRING, `char_value` STRING, `binary_value` BINARY, `varbinary_value` BINARY, `date_value` DATE, `timestamp_value` TIMESTAMP_NTZ, `timestamp_ltz_value` TIMESTAMP_LTZ, `string_array` ARRAY, `string_to_int` MAP, `attributes` STRUCT<`SOURCE`: STRING, `PRIORITY`: INTEGER>, PRIMARY KEY (`id`)); +CREATE TABLE IF NOT EXISTS `SparkTypeSource` (`id` BIGINT NOT NULL, `tiny_value` TINYINT, `small_value` SMALLINT, `int_value` INTEGER, `float_value` FLOAT, `double_value` DOUBLE, `decimal_value` DECIMAL(12, 2), `boolean_value` BOOLEAN, `string_value` STRING, `varchar_value` STRING, `char_value` STRING, `binary_value` BINARY, `varbinary_value` BINARY, `date_value` DATE, `timestamp_value` TIMESTAMP_NTZ, `timestamp_ltz_value` TIMESTAMP_LTZ, `string_array` ARRAY, `string_to_int` MAP, `attributes` STRUCT<`SOURCE`: STRING, `PRIORITY`: INTEGER>, PRIMARY KEY (`id`)) diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/dialects/spark-sql-compile/package.json b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/dialects/spark-sql-compile/package.json new file mode 100644 index 000000000..b229a6399 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/dialects/spark-sql-compile/package.json @@ -0,0 +1,17 @@ +{ + "version": "1", + "enabled-engines": ["flink", "iceberg", "sparksql"], + "script": { + "main": "spark-sql-dialect.sqrl" + }, + "engines": { + "flink": { + "config": { + "table.exec.source.idle-timeout": "1 ms" + } + } + }, + "test-runner": { + "delay-sec": -1 + } +} diff --git a/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/dialects/spark-sql-compile/spark-sql-dialect.sqrl b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/dialects/spark-sql-compile/spark-sql-dialect.sqrl new file mode 100644 index 000000000..1a15b0241 --- /dev/null +++ b/sqrl-testing/sqrl-testing-integration/src/test/resources/usecases/dialects/spark-sql-compile/spark-sql-dialect.sqrl @@ -0,0 +1,56 @@ +/*+engine(iceberg), no_query */ +CREATE TABLE SparkTypeSource ( + id BIGINT NOT NULL, + tiny_value TINYINT, + small_value SMALLINT, + int_value INT, + float_value FLOAT, + double_value DOUBLE, + decimal_value DECIMAL(12, 2), + boolean_value BOOLEAN, + string_value STRING, + varchar_value VARCHAR(32), + char_value CHAR(8), + binary_value BINARY(16), + varbinary_value VARBINARY(32), + date_value DATE, + timestamp_value TIMESTAMP(3), + timestamp_ltz_value TIMESTAMP_LTZ(3), + string_array ARRAY, + string_to_int MAP, + attributes ROW, + PRIMARY KEY (id) NOT ENFORCED +); + +SparkTypeMappings := + SELECT + id, + tiny_value, + small_value, + int_value, + float_value, + double_value, + decimal_value, + boolean_value, + string_value, + varchar_value, + char_value, + binary_value, + varbinary_value, + date_value, + timestamp_value, + timestamp_ltz_value, + string_array, + string_to_int, + attributes + FROM SparkTypeSource; + +SparkAggregates := + SELECT + boolean_value, + COUNT(*) AS record_count, + SUM(decimal_value) AS decimal_sum, + AVG(double_value) AS double_average, + MAX(timestamp_ltz_value) AS latest_event + FROM SparkTypeSource + GROUP BY boolean_value;