This is an automated email from the ASF dual-hosted git repository.
hansva pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new cdff02ffdf Issue #7544 : Add a native Spark SQL transform (#7859)
cdff02ffdf is described below
commit cdff02ffdf8cd4d093fb50d6c70fc9f4199b3fc6
Author: vbhanuchander-lang <[email protected]>
AuthorDate: Thu Aug 13 03:10:30 2026 -0400
Issue #7544 : Add a native Spark SQL transform (#7859)
* Issue #7544 : Add a native Spark SQL transform
Adds a Spark SQL transform for the native Spark pipeline engine. Each
incoming transform is registered as a temporary view, so one statement
can join, union, filter, window and aggregate them under a single
Catalyst plan instead of chaining Hop transforms that Catalyst cannot
optimize across.
Unlike the classic SQL transforms (Execute SQL script, Table input,
Dynamic SQL row) this compiles SQL against the pipeline's in-flight
Datasets rather than sending it to an RDBMS over JDBC, so no database
connection is involved.
- Multi-input: one temp view per enabled incoming hop, named after the
transform (sanitized to a SQL-safe identifier) or by an explicit
per-input override. Colliding view names are reported rather than
silently shadowing each other.
- Zero-input statements are allowed for catalog tables and VALUES.
- Hop variables are resolved driver-side before the statement runs.
- The declared output field list drives the projection: fields are
selected in declared order and cast to the Spark type for their Hop
type, so the Dataset schema matches the row layout getFields()
reports at design time. A non-empty list is required in v1.
lookupPreviousDataset() becomes public so the handler reuses the
converter's target-stream rules instead of duplicating them.
Includes unit tests on a local SparkSession covering joins, unions
across partitions, view naming and overrides, zero-input statements,
variable substitution, declared-schema projection and the error paths,
plus user and dev manual documentation.
* Issue #7544 : Fix the javadoc reference that broke the build
The class javadoc linked {@link #getViews()}, which Lombok generates from
the @Getter on the field, so javadoc cannot resolve it:
SparkSqlMeta.java:92: error: reference not found
maven-javadoc-plugin fails the build on that, so the Code job went red
while the Documentation job passed. Replaced the link with a {@code views}
reference, which needs no resolution.
javadoc:javadoc on plugins/engines/spark is green again, and the module's
Spark SQL tests still pass (18).
---
.../assets/images/transforms/icons/spark-sql.svg | 5 +
docs/hop-user-manual/modules/ROOT/nav.adoc | 1 +
.../spark/getting-started-with-native-spark.adoc | 5 +
.../modules/ROOT/pages/pipeline/transforms.adoc | 1 +
.../ROOT/pages/pipeline/transforms/spark-sql.adoc | 132 ++++++++
.../pipeline/HopPipelineMetaToSparkConverter.java | 10 +-
.../spark/pipeline/handler/SparkSqlHandler.java | 254 ++++++++++++++
.../apache/hop/spark/transforms/sql/SparkSql.java | 48 +++
.../hop/spark/transforms/sql/SparkSqlData.java | 27 ++
.../hop/spark/transforms/sql/SparkSqlDialog.java | 289 ++++++++++++++++
.../hop/spark/transforms/sql/SparkSqlMeta.java | 148 ++++++++
.../hop/spark/transforms/sql/SparkSqlView.java | 48 +++
.../java/org/apache/hop/spark/util/SparkConst.java | 3 +
.../sql/messages/messages_en_US.properties | 34 ++
.../engines/spark/src/main/resources/spark-sql.svg | 5 +
.../pipeline/handler/SparkSqlHandlerTest.java | 372 +++++++++++++++++++++
.../hop/spark/transforms/sql/SparkSqlMetaTest.java | 106 ++++++
17 files changed, 1485 insertions(+), 3 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/spark-sql.svg
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/spark-sql.svg
new file mode 100644
index 0000000000..24990f42cb
--- /dev/null
+++
b/docs/hop-user-manual/modules/ROOT/assets/images/transforms/icons/spark-sql.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,4 H 22 V 20 H 2 Z M 4,6 V 9 H 20 V 6 Z M 4,11 v 3 h 7 v -3 z
M 13,11 v 3 h 7 v -3 z M 4,16 v 3 h 7 v -3 z M 13,16 v 3 h 7 v -3 z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git a/docs/hop-user-manual/modules/ROOT/nav.adoc
b/docs/hop-user-manual/modules/ROOT/nav.adoc
index 3370f79813..b02455a43e 100644
--- a/docs/hop-user-manual/modules/ROOT/nav.adoc
+++ b/docs/hop-user-manual/modules/ROOT/nav.adoc
@@ -268,6 +268,7 @@ under the License.
*** xref:pipeline/transforms/spark-lake-table-maintenance.adoc[Spark lake
table maintenance]
*** xref:pipeline/transforms/spark-lake-table-merge.adoc[Spark lake table
merge]
*** xref:pipeline/transforms/spark-lake-table-output.adoc[Spark lake table
output]
+*** xref:pipeline/transforms/spark-sql.adoc[Spark SQL]
*** xref:pipeline/transforms/splitfields.adoc[Split Fields]
*** xref:pipeline/transforms/splitfieldtorows.adoc[Split fields to rows]
*** xref:pipeline/transforms/splunkinput.adoc[Splunk input]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc
index f0fa23c38f..4e01a73626 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/spark/getting-started-with-native-spark.adoc
@@ -546,8 +546,13 @@ These use Spark Dataset APIs on the driver graph (no
per-row Hop mini-pipeline f
|Spark file input / output
|`spark.read` / `df.write`
+
+|xref:pipeline/transforms/spark-sql.adoc[Spark SQL]
+|`spark.sql` over the incoming Datasets, each registered as a temporary view
|===
+TIP: When a chain of Join / Filter / Calculator transforms exists only to
express analytic logic, a single xref:pipeline/transforms/spark-sql.adoc[Spark
SQL] transform lets Catalyst plan the whole thing as one query instead.
+
=== Generic mapPartitions transforms
Most other transforms — Filter Rows, Switch/Case, Stream Lookup, Calculator,
Select Values, Dummy, classic file transforms, Excel, and many plugins — run
inside `mapPartitions`:
diff --git a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
index c120097ef5..3e1a4b1064 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms.adoc
@@ -216,6 +216,7 @@ The pages nested under this topic contain information on
how to use the transfor
* xref:pipeline/transforms/spark-lake-table-maintenance.adoc[Spark lake table
maintenance]
* xref:pipeline/transforms/spark-lake-table-merge.adoc[Spark lake table merge]
* xref:pipeline/transforms/spark-lake-table-output.adoc[Spark lake table
output]
+* xref:pipeline/transforms/spark-sql.adoc[Spark SQL]
* xref:pipeline/transforms/splitfields.adoc[Split Fields]
* xref:pipeline/transforms/splitfieldtorows.adoc[Split fields to rows]
* xref:pipeline/transforms/splunkinput.adoc[Splunk input]
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-sql.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-sql.adoc
new file mode 100644
index 0000000000..c3c04303de
--- /dev/null
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/spark-sql.adoc
@@ -0,0 +1,132 @@
+////
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you 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.
+////
+:documentationPath: /pipeline/transforms/
+:language: en_US
+:description: Spark SQL runs a SQL statement over the Datasets of the incoming
transforms on the native Spark pipeline engine, so joins, unions and
aggregations are planned as one Catalyst query.
+
+= image:transforms/icons/spark-sql.svg[Spark SQL transform Icon,
role="image-doc-icon"] Spark SQL
+
+[%noheader,cols="3a,1a", role="table-no-borders" ]
+|===
+|
+== Description
+
+*Spark SQL* runs a https://spark.apache.org/sql/[Spark SQL] statement on the
xref:pipeline/spark/getting-started-with-native-spark.adoc[native Spark
pipeline engine].
+
+Every incoming transform is registered as a temporary view, so a single
statement can join, union, filter, window and aggregate them. Spark's Catalyst
optimizer plans the whole statement at once, rather than the pipeline chaining
separate Hop Join, Filter and Calculator transforms that Catalyst cannot see
across.
+
+This is *not* the same as the classic
xref:pipeline/transforms/execsqlrow.adoc[Execute SQL script],
xref:pipeline/transforms/tableinput.adoc[Table input] or
xref:pipeline/transforms/dynamicsqlrow.adoc[Dynamic SQL row] transforms: those
send SQL to a relational database over JDBC. Spark SQL compiles the statement
against the pipeline's in-flight Datasets instead, and needs no database
connection.
+
+This transform is *not* available on the local Hop engine or on Beam engines.
+
+|
+== Supported Engines
+[%noheader,cols="2,1a",frame=none, role="table-supported-engines"]
+!===
+!Hop Engine! image:cross.svg[Not Supported, 24]
+!Single Threaded! image:cross.svg[Not Supported, 24]
+!Native Spark! image:check_mark.svg[Supported, 24]
+!Beam Spark! image:cross.svg[Not Supported, 24]
+!Beam Flink! image:cross.svg[Not Supported, 24]
+!Beam Dataflow! image:cross.svg[Not Supported, 24]
+!===
+
+NOTE: *Spark* here means Hop's *native* Spark pipeline engine only (not Beam
Spark).
+|===
+
+== Options
+
+[options="header",cols="1,3"]
+|===
+|Option |Description
+
+|Transform name
+|Unique name of the transform in the pipeline.
+
+|SQL statement
+|The statement passed to `SparkSession.sql()`. Hop variables (`${...}`) are
resolved on the driver before the statement runs — see <<Variables>>.
+
+|Input views
+|Optional per-input overrides of the temporary view name. Leave empty to use
the default derived from each incoming transform's name — see <<Input views>>.
+
+|Output fields
+|The fields the statement returns. Required — see <<Output fields>>.
+|===
+
+== Input views
+
+Each enabled incoming hop contributes one temporary view. By default the view
name is derived from the incoming transform name: every character outside
`A-Z`, `a-z`, `0-9` and `_` becomes an underscore, and a leading digit is
prefixed with an underscore.
+
+For example, a transform called `Read orders (raw)` is available as
`Read_orders__raw_`.
+
+Because that is rarely what you want to type in a statement, set an explicit
name in the *Input views* grid instead:
+
+[options="header",cols="1,1"]
+|===
+|Incoming transform |View name
+
+|Read orders (raw)
+|orders
+
+|Read customers
+|customers
+|===
+
+The statement can then read naturally:
+
+[source,sql]
+----
+SELECT c.name AS name,
+ SUM(o.amount) AS total
+FROM orders o
+JOIN customers c ON o.customer_id = c.id
+GROUP BY c.name
+----
+
+If two incoming transforms resolve to the same view name, the pipeline fails
at build time with both transform names, rather than one silently shadowing the
other.
+
+A statement with no incoming hops is allowed, so catalog tables and literals
work too:
+
+[source,sql]
+----
+SELECT * FROM VALUES (1, 'a'), (2, 'b') AS t(id, label)
+----
+
+== Output fields
+
+The output field list is required and must not be empty. Hop resolves field
names and types at design time, before any Spark session exists, so the
statement's result schema is not available to it. Declaring the fields keeps
design-time metadata correct and gives downstream transforms a row layout that
matches the Dataset exactly.
+
+The declared list also drives the output: fields are selected in the order you
declare them and cast to the Spark type for the Hop type you choose. Declaring
a field the statement does not return fails the pipeline with the list of
columns it did return.
+
+TIP: Alias the expressions in the statement (`SUM(o.amount) AS total`) so the
returned column names match the fields you declare.
+
+== Variables
+
+Variables are substituted into the SQL text on the driver before the statement
is parsed — the result is plain string substitution, not a bound parameter:
+
+[source,sql]
+----
+SELECT COUNT(*) AS row_count FROM orders WHERE amount >= ${MIN_AMOUNT}
+----
+
+WARNING: Because this is string substitution, a variable whose value comes
from outside the project can change the meaning of the statement. Only use
variables you control for SQL text.
+
+== Notes and limitations
+
+* Structured Streaming SQL is not supported: the native Spark engine is
batch-only.
+* Views exist only for the statement that reads them. They are not visible to
other transforms or to later pipelines.
+* `CREATE`/`DROP` statements against a configured catalog run, but this
transform is designed to return a result set; use
xref:pipeline/transforms/spark-lake-table-maintenance.adoc[Spark lake table
maintenance] for table maintenance operations.
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java
index 8d16d7e304..76298a394c 100644
---
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java
@@ -52,6 +52,7 @@ import
org.apache.hop.spark.pipeline.handler.SparkLakeTableOutputHandler;
import org.apache.hop.spark.pipeline.handler.SparkMemoryGroupByHandler;
import org.apache.hop.spark.pipeline.handler.SparkMergeJoinHandler;
import org.apache.hop.spark.pipeline.handler.SparkSortRowsHandler;
+import org.apache.hop.spark.pipeline.handler.SparkSqlHandler;
import org.apache.hop.spark.pipeline.handler.SparkUniqueRowsHandler;
import org.apache.hop.spark.util.SparkConst;
import org.apache.spark.sql.Dataset;
@@ -80,7 +81,8 @@ public class HopPipelineMetaToSparkConverter {
SparkConst.SPARK_LAKE_TABLE_INPUT_PLUGIN_ID,
SparkConst.SPARK_LAKE_TABLE_OUTPUT_PLUGIN_ID,
SparkConst.SPARK_LAKE_TABLE_MERGE_PLUGIN_ID,
- SparkConst.SPARK_LAKE_TABLE_MAINTENANCE_PLUGIN_ID);
+ SparkConst.SPARK_LAKE_TABLE_MAINTENANCE_PLUGIN_ID,
+ SparkConst.SPARK_SQL_PLUGIN_ID);
/**
* Plugin ids that must not run as partition-local Hop mini-pipelines. Keep
in lockstep with
@@ -170,6 +172,7 @@ public class HopPipelineMetaToSparkConverter {
SparkConst.SPARK_LAKE_TABLE_MERGE_PLUGIN_ID, new
SparkLakeTableMergeHandler());
transformHandlers.put(
SparkConst.SPARK_LAKE_TABLE_MAINTENANCE_PLUGIN_ID, new
SparkLakeTableMaintenanceHandler());
+ transformHandlers.put(SparkConst.SPARK_SQL_PLUGIN_ID, new
SparkSqlHandler());
}
public void validatePipeline() throws HopException {
@@ -363,9 +366,10 @@ public class HopPipelineMetaToSparkConverter {
/**
* Prefer a target-stream Dataset when {@code previous} routed to {@code
current} (Filter/Switch);
- * otherwise use the previous transform's main Dataset.
+ * otherwise use the previous transform's main Dataset. Public so native
handlers that resolve
+ * their own named inputs (Spark SQL) reuse the same target-stream rules as
the converter.
*/
- static Dataset<Row> lookupPreviousDataset(
+ public static Dataset<Row> lookupPreviousDataset(
Map<String, Dataset<Row>> transformDatasetMap,
TransformMeta previous,
TransformMeta current,
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkSqlHandler.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkSqlHandler.java
new file mode 100644
index 0000000000..09734456a1
--- /dev/null
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkSqlHandler.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.pipeline.handler;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILogChannel;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.spark.core.HopSparkRowConverter;
+import org.apache.hop.spark.core.SparkNativeMetrics;
+import org.apache.hop.spark.engines.ISparkPipelineEngineRunConfiguration;
+import org.apache.hop.spark.pipeline.HopPipelineMetaToSparkConverter;
+import org.apache.hop.spark.transforms.io.SparkField;
+import org.apache.hop.spark.transforms.sql.SparkSqlMeta;
+import org.apache.spark.sql.Column;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.DataType;
+
+/**
+ * Native Spark SQL execution. Each incoming transform's Dataset is registered
as a temporary view
+ * so a single statement can join, union and aggregate them under one Catalyst
plan.
+ *
+ * <p>The statement result is projected onto the transform's declared field
list so the Dataset
+ * schema matches the Hop row layout that {@link SparkSqlMeta#getFields}
reports at design time.
+ */
+public class SparkSqlHandler extends SparkBaseTransformHandler {
+
+ @Override
+ public void handleTransform(
+ ILogChannel log,
+ IVariables variables,
+ String runConfigurationName,
+ ISparkPipelineEngineRunConfiguration runConfiguration,
+ IHopMetadataProvider metadataProvider,
+ String metastoreJson,
+ PipelineMeta pipelineMeta,
+ TransformMeta transformMeta,
+ Map<String, Dataset<Row>> transformDatasetMap,
+ SparkSession spark,
+ IRowMeta rowMeta,
+ List<TransformMeta> previousTransforms,
+ Dataset<Row> input)
+ throws HopException {
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ loadTransformMetadata(meta, transformMeta, metadataProvider, pipelineMeta);
+
+ String transformName = transformMeta.getName();
+ String sql = variables.resolve(meta.getSql());
+ if (StringUtils.isBlank(sql)) {
+ throw new HopException("Spark SQL '" + transformName + "': the SQL
statement is empty");
+ }
+
+ List<SparkField> fields = meta.getFields();
+ if (fields == null || fields.isEmpty()) {
+ throw new HopException(
+ "Spark SQL '"
+ + transformName
+ + "': an explicit output field list is required so Hop can
resolve the row layout"
+ + " at design time. Add the fields the statement returns.");
+ }
+
+ Map<String, String> registeredViews =
+ registerInputViews(
+ log, variables, meta, transformMeta, previousTransforms,
transformDatasetMap);
+
+ Dataset<Row> result;
+ try {
+ result = spark.sql(sql);
+ } catch (Exception e) {
+ throw new HopException(
+ "Spark SQL '"
+ + transformName
+ + "': the statement could not be executed. Registered views: "
+ + registeredViews.values()
+ + ". SQL: "
+ + sql,
+ e);
+ }
+
+ Dataset<Row> output = projectDeclaredFields(result, fields, transformName);
+ output = trackMetrics(output, transformMeta,
SparkNativeMetrics.Role.TRANSFORM);
+ transformDatasetMap.put(transformName, output);
+
+ log.logBasic(
+ "Handled Spark SQL : "
+ + transformName
+ + " views="
+ + registeredViews
+ + " columns="
+ + Arrays.toString(output.columns()));
+ }
+
+ /**
+ * Registers one temporary view per incoming transform and returns the
transform-name to view-name
+ * mapping. An empty map means a zero-input statement (catalog tables,
{@code VALUES}, …).
+ */
+ private Map<String, String> registerInputViews(
+ ILogChannel log,
+ IVariables variables,
+ SparkSqlMeta meta,
+ TransformMeta transformMeta,
+ List<TransformMeta> previousTransforms,
+ Map<String, Dataset<Row>> transformDatasetMap)
+ throws HopException {
+
+ Map<String, String> registered = new LinkedHashMap<>();
+ if (previousTransforms == null || previousTransforms.isEmpty()) {
+ return registered;
+ }
+
+ // view name -> source transform, to report collisions rather than
silently shadowing
+ Map<String, String> viewOwners = new LinkedHashMap<>();
+
+ for (TransformMeta previous : previousTransforms) {
+ String previousName = previous.getName();
+ Dataset<Row> dataset =
+ HopPipelineMetaToSparkConverter.lookupPreviousDataset(
+ transformDatasetMap, previous, transformMeta, log);
+ if (dataset == null) {
+ throw new HopException(
+ "Spark SQL '"
+ + transformMeta.getName()
+ + "': the Dataset for incoming transform '"
+ + previousName
+ + "' could not be found. Check that the hop into this
transform is enabled.");
+ }
+
+ String override = meta.findViewNameOverride(previousName);
+ String viewName =
+ StringUtils.isNotEmpty(override)
+ ? variables.resolve(override)
+ : SparkSqlMeta.defaultViewName(previousName);
+ if (StringUtils.isBlank(viewName)) {
+ throw new HopException(
+ "Spark SQL '"
+ + transformMeta.getName()
+ + "': could not determine a view name for incoming transform '"
+ + previousName
+ + "'");
+ }
+
+ String owner = viewOwners.put(viewName, previousName);
+ if (owner != null) {
+ throw new HopException(
+ "Spark SQL '"
+ + transformMeta.getName()
+ + "': incoming transforms '"
+ + owner
+ + "' and '"
+ + previousName
+ + "' both map to view '"
+ + viewName
+ + "'. Set distinct view names in the Input views tab.");
+ }
+
+ dataset.createOrReplaceTempView(viewName);
+ registered.put(previousName, viewName);
+ }
+ return registered;
+ }
+
+ /**
+ * Selects the declared fields, in declared order, casting each to the Spark
type for its Hop
+ * type. Guarantees the Dataset schema matches the row layout reported at
design time.
+ */
+ private Dataset<Row> projectDeclaredFields(
+ Dataset<Row> result, List<SparkField> fields, String transformName)
throws HopException {
+
+ String[] resultColumns = result.columns();
+ List<Column> projection = new ArrayList<>(fields.size());
+
+ for (SparkField field : fields) {
+ String name = field.getName();
+ if (StringUtils.isBlank(name)) {
+ throw new HopException(
+ "Spark SQL '" + transformName + "': a declared output field has no
name");
+ }
+
+ String actual = findColumn(resultColumns, name);
+ if (actual == null) {
+ throw new HopException(
+ "Spark SQL '"
+ + transformName
+ + "': declared output field '"
+ + name
+ + "' is not returned by the statement. Columns returned: "
+ + Arrays.toString(resultColumns));
+ }
+
+ IValueMeta valueMeta;
+ try {
+ valueMeta = field.createValueMeta();
+ } catch (Exception e) {
+ throw new HopException(
+ "Spark SQL '"
+ + transformName
+ + "': unable to build the value metadata for output field '"
+ + name
+ + "'",
+ e);
+ }
+ DataType dataType = HopSparkRowConverter.toDataType(valueMeta);
+ projection.add(result.col(actual).cast(dataType).alias(name));
+ }
+
+ return result.select(projection.toArray(new Column[0]));
+ }
+
+ /**
+ * Resolves a declared field against the statement's columns, preferring an
exact match and
+ * falling back to a case-insensitive one (Spark resolves case-insensitively
by default).
+ */
+ private static String findColumn(String[] resultColumns, String name) {
+ for (String column : resultColumns) {
+ if (column.equals(name)) {
+ return column;
+ }
+ }
+ for (String column : resultColumns) {
+ if (column.equalsIgnoreCase(name)) {
+ return column;
+ }
+ }
+ return null;
+ }
+}
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSql.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSql.java
new file mode 100644
index 0000000000..95f92b4f01
--- /dev/null
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSql.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.transforms.sql;
+
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.pipeline.Pipeline;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transform.BaseTransform;
+import org.apache.hop.pipeline.transform.TransformMeta;
+
+/**
+ * Metadata-only on the Local engine. On the native Spark engine {@link
+ * org.apache.hop.spark.pipeline.handler.SparkSqlHandler} compiles this into a
{@code
+ * SparkSession.sql()} call over the incoming Datasets.
+ */
+public class SparkSql extends BaseTransform<SparkSqlMeta, SparkSqlData> {
+
+ public SparkSql(
+ TransformMeta transformMeta,
+ SparkSqlMeta meta,
+ SparkSqlData data,
+ int copyNr,
+ PipelineMeta pipelineMeta,
+ Pipeline pipeline) {
+ super(transformMeta, meta, data, copyNr, pipelineMeta, pipeline);
+ }
+
+ @Override
+ public boolean processRow() throws HopException {
+ setOutputDone();
+ return false;
+ }
+}
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlData.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlData.java
new file mode 100644
index 0000000000..3e520d45a4
--- /dev/null
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlData.java
@@ -0,0 +1,27 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.transforms.sql;
+
+import org.apache.hop.pipeline.transform.BaseTransformData;
+import org.apache.hop.pipeline.transform.ITransformData;
+
+public class SparkSqlData extends BaseTransformData implements ITransformData {
+ public SparkSqlData() {
+ super();
+ }
+}
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlDialog.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlDialog.java
new file mode 100644
index 0000000000..2d69734e71
--- /dev/null
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlDialog.java
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.transforms.sql;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hop.core.Const;
+import org.apache.hop.core.row.value.ValueMetaFactory;
+import org.apache.hop.core.util.Utils;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.i18n.BaseMessages;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.spark.transforms.io.SparkField;
+import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.dialog.BaseDialog;
+import org.apache.hop.ui.core.widget.ColumnInfo;
+import org.apache.hop.ui.core.widget.TableView;
+import org.apache.hop.ui.pipeline.transform.BaseTransformDialog;
+import org.eclipse.swt.SWT;
+import org.eclipse.swt.layout.FormAttachment;
+import org.eclipse.swt.layout.FormData;
+import org.eclipse.swt.layout.FormLayout;
+import org.eclipse.swt.widgets.Button;
+import org.eclipse.swt.widgets.Control;
+import org.eclipse.swt.widgets.Label;
+import org.eclipse.swt.widgets.Shell;
+import org.eclipse.swt.widgets.TableItem;
+import org.eclipse.swt.widgets.Text;
+
+public class SparkSqlDialog extends BaseTransformDialog {
+ private static final Class<?> PKG = SparkSqlMeta.class;
+
+ private final SparkSqlMeta input;
+
+ private Text wSql;
+ private TableView wViews;
+ private TableView wFields;
+
+ public SparkSqlDialog(
+ Shell parent, IVariables variables, SparkSqlMeta transformMeta,
PipelineMeta pipelineMeta) {
+ super(parent, variables, transformMeta, pipelineMeta);
+ this.input = transformMeta;
+ }
+
+ @Override
+ public String open() {
+ Shell parent = getParent();
+ shell = new Shell(parent, SWT.DIALOG_TRIM | SWT.RESIZE | SWT.MIN |
SWT.MAX);
+ PropsUi.setLook(shell);
+ setShellImage(shell, input);
+
+ changed = input.hasChanged();
+
+ FormLayout formLayout = new FormLayout();
+ formLayout.marginWidth = PropsUi.getFormMargin();
+ formLayout.marginHeight = PropsUi.getFormMargin();
+ shell.setLayout(formLayout);
+ shell.setText(BaseMessages.getString(PKG, "SparkSqlDialog.Shell.Title"));
+
+ int middle = props.getMiddlePct();
+ int margin = PropsUi.getMargin();
+
+ wlTransformName = new Label(shell, SWT.RIGHT);
+ wlTransformName.setText(BaseMessages.getString(PKG,
"System.Label.TransformName"));
+ PropsUi.setLook(wlTransformName);
+ fdlTransformName = new FormData();
+ fdlTransformName.left = new FormAttachment(0, 0);
+ fdlTransformName.top = new FormAttachment(0, margin);
+ fdlTransformName.right = new FormAttachment(middle, -margin);
+ wlTransformName.setLayoutData(fdlTransformName);
+ wTransformName = new Text(shell, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
+ wTransformName.setText(transformName);
+ PropsUi.setLook(wTransformName);
+ wTransformName.addModifyListener(e -> input.setChanged());
+ fdTransformName = new FormData();
+ fdTransformName.left = new FormAttachment(middle, 0);
+ fdTransformName.top = new FormAttachment(wlTransformName, 0, SWT.CENTER);
+ fdTransformName.right = new FormAttachment(100, 0);
+ wTransformName.setLayoutData(fdTransformName);
+ Control last = wTransformName;
+
+ Label wlSql = new Label(shell, SWT.LEFT);
+ wlSql.setText(BaseMessages.getString(PKG, "SparkSqlDialog.Sql"));
+ PropsUi.setLook(wlSql);
+ FormData fdlSql = new FormData();
+ fdlSql.left = new FormAttachment(0, 0);
+ fdlSql.top = new FormAttachment(last, margin);
+ fdlSql.right = new FormAttachment(100, 0);
+ wlSql.setLayoutData(fdlSql);
+
+ wSql = new Text(shell, SWT.MULTI | SWT.LEFT | SWT.BORDER | SWT.V_SCROLL |
SWT.H_SCROLL);
+ PropsUi.setLook(wSql);
+ wSql.addModifyListener(e -> input.setChanged());
+ wSql.setToolTipText(BaseMessages.getString(PKG,
"SparkSqlDialog.Sql.Tooltip"));
+ FormData fdSql = new FormData();
+ fdSql.left = new FormAttachment(0, 0);
+ fdSql.top = new FormAttachment(wlSql, margin);
+ fdSql.right = new FormAttachment(100, 0);
+ fdSql.height = (int) (150 * props.getZoomFactor());
+ wSql.setLayoutData(fdSql);
+ last = wSql;
+
+ Label wlViews = new Label(shell, SWT.LEFT);
+ wlViews.setText(BaseMessages.getString(PKG, "SparkSqlDialog.Views"));
+ PropsUi.setLook(wlViews);
+ FormData fdlViews = new FormData();
+ fdlViews.left = new FormAttachment(0, 0);
+ fdlViews.top = new FormAttachment(last, margin);
+ fdlViews.right = new FormAttachment(100, 0);
+ wlViews.setLayoutData(fdlViews);
+
+ String[] previousTransforms =
pipelineMeta.getPrevTransformNames(transformName);
+ ColumnInfo[] viewColumns =
+ new ColumnInfo[] {
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "SparkSqlDialog.Column.Transform"),
+ ColumnInfo.COLUMN_TYPE_CCOMBO,
+ previousTransforms == null ? new String[0] : previousTransforms,
+ false),
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "SparkSqlDialog.Column.View"),
+ ColumnInfo.COLUMN_TYPE_TEXT,
+ false)
+ };
+ wViews =
+ new TableView(
+ variables,
+ shell,
+ SWT.BORDER | SWT.FULL_SELECTION | SWT.MULTI,
+ viewColumns,
+ input.getViews() == null ? 1 : Math.max(1,
input.getViews().size()),
+ e -> input.setChanged(),
+ props);
+ FormData fdViews = new FormData();
+ fdViews.left = new FormAttachment(0, 0);
+ fdViews.top = new FormAttachment(wlViews, margin);
+ fdViews.right = new FormAttachment(100, 0);
+ fdViews.height = (int) (100 * props.getZoomFactor());
+ wViews.setLayoutData(fdViews);
+ last = wViews;
+
+ Label wlFields = new Label(shell, SWT.LEFT);
+ wlFields.setText(BaseMessages.getString(PKG, "SparkSqlDialog.Fields"));
+ PropsUi.setLook(wlFields);
+ FormData fdlFields = new FormData();
+ fdlFields.left = new FormAttachment(0, 0);
+ fdlFields.top = new FormAttachment(last, margin);
+ fdlFields.right = new FormAttachment(100, 0);
+ wlFields.setLayoutData(fdlFields);
+
+ ColumnInfo[] fieldColumns =
+ new ColumnInfo[] {
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "SparkSqlDialog.Column.Name"),
+ ColumnInfo.COLUMN_TYPE_TEXT,
+ false),
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "SparkSqlDialog.Column.Type"),
+ ColumnInfo.COLUMN_TYPE_CCOMBO,
+ ValueMetaFactory.getValueMetaNames()),
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "SparkSqlDialog.Column.Length"),
+ ColumnInfo.COLUMN_TYPE_TEXT,
+ false),
+ new ColumnInfo(
+ BaseMessages.getString(PKG, "SparkSqlDialog.Column.Precision"),
+ ColumnInfo.COLUMN_TYPE_TEXT,
+ false)
+ };
+ wFields =
+ new TableView(
+ variables,
+ shell,
+ SWT.BORDER | SWT.FULL_SELECTION | SWT.MULTI,
+ fieldColumns,
+ input.getFields() == null ? 1 : Math.max(1,
input.getFields().size()),
+ e -> input.setChanged(),
+ props);
+ FormData fdFields = new FormData();
+ fdFields.left = new FormAttachment(0, 0);
+ fdFields.top = new FormAttachment(wlFields, margin);
+ fdFields.right = new FormAttachment(100, 0);
+ fdFields.bottom = new FormAttachment(100, -50);
+ wFields.setLayoutData(fdFields);
+
+ wOk = new Button(shell, SWT.PUSH);
+ wOk.setText(BaseMessages.getString(PKG, "System.Button.OK"));
+ wCancel = new Button(shell, SWT.PUSH);
+ wCancel.setText(BaseMessages.getString(PKG, "System.Button.Cancel"));
+ setButtonPositions(new Button[] {wOk, wCancel}, margin, null);
+
+ wOk.addListener(SWT.Selection, e -> ok());
+ wCancel.addListener(SWT.Selection, e -> cancel());
+
+ getData();
+ input.setChanged(changed);
+ BaseDialog.defaultShellHandling(shell, c -> ok(), c -> cancel());
+ return transformName;
+ }
+
+ private void getData() {
+ wTransformName.setText(Const.NVL(transformName, ""));
+ wSql.setText(Const.NVL(input.getSql(), ""));
+
+ if (input.getViews() != null) {
+ int i = 0;
+ for (SparkSqlView view : input.getViews()) {
+ TableItem item = wViews.table.getItem(i);
+ if (item == null) {
+ item = new TableItem(wViews.table, SWT.NONE);
+ }
+ item.setText(1, Const.NVL(view.getTransformName(), ""));
+ item.setText(2, Const.NVL(view.getViewName(), ""));
+ i++;
+ }
+ wViews.setRowNums();
+ wViews.optWidth(true);
+ }
+
+ if (input.getFields() != null) {
+ int i = 0;
+ for (SparkField f : input.getFields()) {
+ TableItem item = wFields.table.getItem(i);
+ if (item == null) {
+ item = new TableItem(wFields.table, SWT.NONE);
+ }
+ item.setText(1, Const.NVL(f.getName(), ""));
+ item.setText(2, Const.NVL(f.getHopType(), "String"));
+ item.setText(3, f.getLength() >= 0 ? Integer.toString(f.getLength()) :
"");
+ item.setText(4, f.getPrecision() >= 0 ?
Integer.toString(f.getPrecision()) : "");
+ i++;
+ }
+ wFields.setRowNums();
+ wFields.optWidth(true);
+ }
+
+ wTransformName.selectAll();
+ wTransformName.setFocus();
+ }
+
+ private void cancel() {
+ transformName = null;
+ input.setChanged(changed);
+ dispose();
+ }
+
+ private void ok() {
+ if (Utils.isEmpty(wTransformName.getText())) {
+ return;
+ }
+ transformName = wTransformName.getText();
+ input.setSql(wSql.getText());
+
+ List<SparkSqlView> views = new ArrayList<>();
+ for (int i = 0; i < wViews.nrNonEmpty(); i++) {
+ TableItem item = wViews.getNonEmpty(i);
+ views.add(new SparkSqlView(item.getText(1), item.getText(2)));
+ }
+ input.setViews(views);
+
+ List<SparkField> fields = new ArrayList<>();
+ for (int i = 0; i < wFields.nrNonEmpty(); i++) {
+ TableItem item = wFields.getNonEmpty(i);
+ SparkField f = new SparkField();
+ f.setName(item.getText(1));
+ f.setHopType(item.getText(2));
+ f.setLength(Const.toInt(item.getText(3), -1));
+ f.setPrecision(Const.toInt(item.getText(4), -1));
+ fields.add(f);
+ }
+ input.setFields(fields);
+ dispose();
+ }
+}
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlMeta.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlMeta.java
new file mode 100644
index 0000000000..d204c25a70
--- /dev/null
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlMeta.java
@@ -0,0 +1,148 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.transforms.sql;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.core.annotations.Transform;
+import org.apache.hop.core.exception.HopPluginException;
+import org.apache.hop.core.exception.HopTransformException;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.pipeline.transform.BaseTransformMeta;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.spark.transforms.io.SparkField;
+import org.apache.hop.spark.util.SparkConst;
+
+/**
+ * Runs a Spark SQL statement against the Datasets of the incoming transforms,
each registered as a
+ * temporary view. Native Spark engine only.
+ *
+ * <p>The output row layout is not inferred from the statement: v1 requires an
explicit, non-empty
+ * field list so that Hop can resolve fields at design time and so downstream
generic mapPartitions
+ * transforms see a row layout that matches the Dataset exactly.
+ */
+@Transform(
+ id = SparkConst.SPARK_SQL_PLUGIN_ID,
+ name = "i18n::SparkSql.Name",
+ description = "i18n::SparkSql.Description",
+ image = "spark-sql.svg",
+ categoryDescription =
"i18n:org.apache.hop.pipeline.transform:BaseTransform.Category.BigData",
+ keywords = "i18n::SparkSql.Keyword",
+ documentationUrl = "/pipeline/transforms/spark-sql.html",
+ supportedEngines = {SparkConst.PLUGIN_ID})
+@Getter
+@Setter
+public class SparkSqlMeta extends BaseTransformMeta<SparkSql, SparkSqlData> {
+
+ /** SQL executed through {@code SparkSession.sql()}. Variables are resolved
driver-side. */
+ @HopMetadataProperty(key = "sql", injectionKey = "SQL")
+ private String sql;
+
+ /** Optional per-input temporary view name overrides. */
+ @HopMetadataProperty(
+ groupKey = "views",
+ key = "view",
+ injectionGroupKey = "VIEWS",
+ injectionGroupDescription = "SparkSql.Injection.Group.Views")
+ private List<SparkSqlView> views = new ArrayList<>();
+
+ /** Output fields. Required and non-empty in v1 — see the class javadoc. */
+ @HopMetadataProperty(
+ groupKey = "fields",
+ key = "field",
+ injectionGroupKey = "FIELDS",
+ injectionGroupDescription = "SparkSql.Injection.Group.Fields")
+ private List<SparkField> fields = new ArrayList<>();
+
+ public SparkSqlMeta() {
+ super();
+ }
+
+ @Override
+ public String getDialogClassName() {
+ return SparkSqlDialog.class.getName();
+ }
+
+ /**
+ * Derives a SQL-safe temporary view name from a transform name: characters
outside {@code
+ * [A-Za-z0-9_]} become underscores, and a leading digit is prefixed with an
underscore. Transform
+ * names such as {@code "Read orders (raw)"} therefore become {@code
"Read_orders__raw_"}.
+ *
+ * <p>Callers that need a stable, readable name should set an explicit
override in the {@code
+ * views} list instead.
+ */
+ public static String defaultViewName(String transformName) {
+ if (transformName == null || transformName.isEmpty()) {
+ return null;
+ }
+ StringBuilder sb = new StringBuilder(transformName.length());
+ for (int i = 0; i < transformName.length(); i++) {
+ char c = transformName.charAt(i);
+ sb.append((Character.isLetterOrDigit(c) && c < 128) || c == '_' ? c :
'_');
+ }
+ if (Character.isDigit(sb.charAt(0))) {
+ sb.insert(0, '_');
+ }
+ return sb.toString();
+ }
+
+ /** Returns the configured view-name override for {@code transformName}, or
null when unset. */
+ public String findViewNameOverride(String transformName) {
+ if (views == null || transformName == null) {
+ return null;
+ }
+ for (SparkSqlView view : views) {
+ if (transformName.equals(view.getTransformName())
+ && view.getViewName() != null
+ && !view.getViewName().isEmpty()) {
+ return view.getViewName();
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public void getFields(
+ IRowMeta inputRowMeta,
+ String name,
+ IRowMeta[] info,
+ TransformMeta nextTransform,
+ IVariables variables,
+ IHopMetadataProvider metadataProvider)
+ throws HopTransformException {
+ // The statement fully determines the output row: incoming fields do not
survive it.
+ inputRowMeta.clear();
+ if (fields == null || fields.isEmpty()) {
+ return;
+ }
+ try {
+ for (SparkField field : fields) {
+ if (field.getName() != null && !field.getName().isEmpty()) {
+ inputRowMeta.addValueMeta(field.createValueMeta());
+ }
+ }
+ } catch (HopPluginException e) {
+ throw new HopTransformException("Unable to create row meta for Spark
SQL", e);
+ }
+ }
+}
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlView.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlView.java
new file mode 100644
index 0000000000..e16f33d793
--- /dev/null
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/transforms/sql/SparkSqlView.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.transforms.sql;
+
+import java.io.Serializable;
+import lombok.Getter;
+import lombok.Setter;
+import org.apache.hop.metadata.api.HopMetadataProperty;
+
+/**
+ * Optional override of the temporary view name used for one incoming
transform. Without an override
+ * the view name is derived from the transform name by {@link
SparkSqlMeta#defaultViewName(String)}.
+ */
+@Getter
+@Setter
+public class SparkSqlView implements Serializable {
+ private static final long serialVersionUID = 1L;
+
+ /** Name of the incoming (previous) transform whose Dataset is registered. */
+ @HopMetadataProperty(key = "transform", injectionKey = "TRANSFORM")
+ private String transformName;
+
+ /** Temporary view name to register the Dataset under. Supports variables. */
+ @HopMetadataProperty(key = "view", injectionKey = "VIEW")
+ private String viewName;
+
+ public SparkSqlView() {}
+
+ public SparkSqlView(String transformName, String viewName) {
+ this.transformName = transformName;
+ this.viewName = viewName;
+ }
+}
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/util/SparkConst.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/util/SparkConst.java
index 92d480ac18..1a55480d62 100644
---
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/util/SparkConst.java
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/util/SparkConst.java
@@ -50,5 +50,8 @@ public final class SparkConst {
public static final String SPARK_LAKE_TABLE_MAINTENANCE_PLUGIN_ID =
"SparkLakeTableMaintenance";
+ /** Spark SQL over the Datasets of the incoming transforms — native Spark
only. */
+ public static final String SPARK_SQL_PLUGIN_ID = "SparkSql";
+
private SparkConst() {}
}
diff --git
a/plugins/engines/spark/src/main/resources/org/apache/hop/spark/transforms/sql/messages/messages_en_US.properties
b/plugins/engines/spark/src/main/resources/org/apache/hop/spark/transforms/sql/messages/messages_en_US.properties
new file mode 100644
index 0000000000..30706123f7
--- /dev/null
+++
b/plugins/engines/spark/src/main/resources/org/apache/hop/spark/transforms/sql/messages/messages_en_US.properties
@@ -0,0 +1,34 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+#
+
+SparkSql.Name=Spark SQL
+SparkSql.Description=Run a Spark SQL statement over the Datasets of the
incoming transforms with the native Spark engine
+SparkSql.Keyword=spark,sql,query,join,union,aggregate,catalyst
+SparkSql.Injection.Group.Views=Input views (optional view name overrides)
+SparkSql.Injection.Group.Fields=Output fields
+
+SparkSqlDialog.Shell.Title=Spark SQL
+SparkSqlDialog.Sql=SQL statement
+SparkSqlDialog.Sql.Tooltip=Executed with SparkSession.sql(). Each incoming
transform is registered as a temporary view. Hop variables (${...}) are
resolved on the driver before the statement runs.
+SparkSqlDialog.Views=Input views (optional — defaults to the incoming
transform name)
+SparkSqlDialog.Fields=Output fields (required)
+SparkSqlDialog.Column.Transform=Incoming transform
+SparkSqlDialog.Column.View=View name
+SparkSqlDialog.Column.Name=Name
+SparkSqlDialog.Column.Type=Type
+SparkSqlDialog.Column.Length=Length
+SparkSqlDialog.Column.Precision=Precision
diff --git a/plugins/engines/spark/src/main/resources/spark-sql.svg
b/plugins/engines/spark/src/main/resources/spark-sql.svg
new file mode 100644
index 0000000000..24990f42cb
--- /dev/null
+++ b/plugins/engines/spark/src/main/resources/spark-sql.svg
@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="utf-8"?>
+<svg xmlns="http://www.w3.org/2000/svg"; width="24" height="24">
+ <path d="M 2,4 H 22 V 20 H 2 Z M 4,6 V 9 H 20 V 6 Z M 4,11 v 3 h 7 v -3 z
M 13,11 v 3 h 7 v -3 z M 4,16 v 3 h 7 v -3 z M 13,16 v 3 h 7 v -3 z"
+ style="fill:#0e3a5a;fill-opacity:1;fill-rule:evenodd"/>
+</svg>
diff --git
a/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/SparkSqlHandlerTest.java
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/SparkSqlHandlerTest.java
new file mode 100644
index 0000000000..6030433c8e
--- /dev/null
+++
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/pipeline/handler/SparkSqlHandlerTest.java
@@ -0,0 +1,372 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.pipeline.handler;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.HopLogStore;
+import org.apache.hop.core.logging.LogChannel;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaInteger;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.IVariables;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.spark.engines.SparkPipelineRunConfiguration;
+import org.apache.hop.spark.transforms.io.SparkField;
+import org.apache.hop.spark.transforms.sql.SparkSqlMeta;
+import org.apache.hop.spark.transforms.sql.SparkSqlView;
+import org.apache.hop.spark.util.SparkConst;
+import org.apache.spark.sql.Dataset;
+import org.apache.spark.sql.Row;
+import org.apache.spark.sql.RowFactory;
+import org.apache.spark.sql.SparkSession;
+import org.apache.spark.sql.types.DataTypes;
+import org.apache.spark.sql.types.StructField;
+import org.apache.spark.sql.types.StructType;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration-style unit tests for the native Spark SQL handler on a local
Spark session. These
+ * prove multi-input view registration, whole-query planning and the
declared-schema contract.
+ */
+class SparkSqlHandlerTest {
+
+ private static SparkSession spark;
+
+ @BeforeAll
+ static void startSpark() throws Exception {
+ HopEnvironment.init();
+ HopLogStore.init();
+ spark =
+ SparkSession.builder()
+ .appName("hop-spark-sql-handler-test")
+ .master("local[2]")
+ .config("spark.ui.enabled", "false")
+ .config("spark.ui.showConsoleProgress", "false")
+ .config("spark.metrics.staticSources.enabled", "false")
+ .config("spark.sql.shuffle.partitions", "4")
+ .config("spark.driver.host", "localhost")
+ .getOrCreate();
+ }
+
+ @AfterAll
+ static void stopSpark() {
+ if (spark != null) {
+ spark.stop();
+ }
+ }
+
+ @Test
+ void joinsTwoInputsRegisteredAsTempViews() throws Exception {
+ Dataset<Row> orders = ordersDataset();
+ Dataset<Row> customers = customersDataset();
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql(
+ "SELECT c.name AS name, SUM(o.amount) AS total "
+ + "FROM orders o JOIN customers c ON o.customer_id = c.id "
+ + "GROUP BY c.name");
+ meta.setFields(List.of(new SparkField("name", "String"), new
SparkField("total", "Integer")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("orders", orders);
+ map.put("customers", customers);
+
+ Dataset<Row> output = run(meta, "sql", map, List.of("orders",
"customers"), new Variables());
+
+ assertArrayEquals(new String[] {"name", "total"}, output.columns());
+ List<Row> rows = output.orderBy("name").collectAsList();
+ assertEquals(2, rows.size());
+ assertEquals("alice", rows.get(0).getString(0));
+ assertEquals(30L, rows.get(0).getLong(1));
+ assertEquals("bob", rows.get(1).getString(0));
+ assertEquals(7L, rows.get(1).getLong(1));
+ }
+
+ @Test
+ void unionsAcrossPartitionsUnderOnePlan() throws Exception {
+ Dataset<Row> orders = ordersDataset().repartition(4);
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql(
+ "SELECT COUNT(*) AS row_count FROM (SELECT * FROM orders UNION ALL
SELECT * FROM orders)");
+ meta.setFields(List.of(new SparkField("row_count", "Integer")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("orders", orders);
+
+ Dataset<Row> output = run(meta, "sql", map, List.of("orders"), new
Variables());
+ assertEquals(8L, output.collectAsList().get(0).getLong(0));
+ }
+
+ @Test
+ void viewNameOverrideIsUsed() throws Exception {
+ Dataset<Row> orders = ordersDataset();
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT SUM(amount) AS total FROM raw_orders");
+ meta.setViews(List.of(new SparkSqlView("Read orders (raw)",
"raw_orders")));
+ meta.setFields(List.of(new SparkField("total", "Integer")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("Read orders (raw)", orders);
+
+ Dataset<Row> output = run(meta, "sql", map, List.of("Read orders (raw)"),
new Variables());
+ assertEquals(37L, output.collectAsList().get(0).getLong(0));
+ }
+
+ @Test
+ void transformNameWithoutOverrideIsSanitisedIntoAViewName() throws Exception
{
+ Dataset<Row> orders = ordersDataset();
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ // "Read orders (raw)" sanitises to "Read_orders__raw_"
+ meta.setSql("SELECT SUM(amount) AS total FROM Read_orders__raw_");
+ meta.setFields(List.of(new SparkField("total", "Integer")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("Read orders (raw)", orders);
+
+ Dataset<Row> output = run(meta, "sql", map, List.of("Read orders (raw)"),
new Variables());
+ assertEquals(37L, output.collectAsList().get(0).getLong(0));
+ }
+
+ @Test
+ void zeroInputStatementIsAllowed() throws Exception {
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT * FROM VALUES (1, 'a'), (2, 'b') AS t(id, label)");
+ meta.setFields(List.of(new SparkField("id", "Integer"), new
SparkField("label", "String")));
+
+ Dataset<Row> output = run(meta, "sql", new HashMap<>(), List.of(), new
Variables());
+
+ assertArrayEquals(new String[] {"id", "label"}, output.columns());
+ assertEquals(2, output.count());
+ }
+
+ @Test
+ void variablesAreResolvedOnTheDriverBeforeExecution() throws Exception {
+ Dataset<Row> orders = ordersDataset();
+
+ IVariables variables = new Variables();
+ variables.setVariable("MIN_AMOUNT", "10");
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT COUNT(*) AS row_count FROM orders WHERE amount >=
${MIN_AMOUNT}");
+ meta.setFields(List.of(new SparkField("row_count", "Integer")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("orders", orders);
+
+ Dataset<Row> output = run(meta, "sql", map, List.of("orders"), variables);
+ assertEquals(2L, output.collectAsList().get(0).getLong(0));
+ }
+
+ @Test
+ void declaredFieldsDriveOutputOrderAndType() throws Exception {
+ Dataset<Row> orders = ordersDataset();
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT customer_id, amount FROM orders");
+ // Declared in the opposite order, and amount declared as String
+ meta.setFields(
+ List.of(new SparkField("amount", "String"), new
SparkField("customer_id", "String")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("orders", orders);
+
+ Dataset<Row> output = run(meta, "sql", map, List.of("orders"), new
Variables());
+
+ assertArrayEquals(new String[] {"amount", "customer_id"},
output.columns());
+ assertEquals(
+ DataTypes.StringType, output.schema().apply("amount").dataType(),
"declared cast applied");
+ assertEquals(DataTypes.StringType,
output.schema().apply("customer_id").dataType());
+ }
+
+ @Test
+ void missingDeclaredFieldFailsWithTheColumnsThatWereReturned() {
+ Dataset<Row> orders = ordersDataset();
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT customer_id FROM orders");
+ meta.setFields(List.of(new SparkField("nope", "String")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("orders", orders);
+
+ HopException e =
+ assertThrows(
+ HopException.class, () -> run(meta, "sql", map, List.of("orders"),
new Variables()));
+ assertTrue(e.getMessage().contains("nope"), e.getMessage());
+ assertTrue(e.getMessage().contains("customer_id"), e.getMessage());
+ }
+
+ @Test
+ void emptyFieldListIsRejected() {
+ Dataset<Row> orders = ordersDataset();
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT customer_id FROM orders");
+ meta.setFields(new ArrayList<>());
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("orders", orders);
+
+ HopException e =
+ assertThrows(
+ HopException.class, () -> run(meta, "sql", map, List.of("orders"),
new Variables()));
+ assertTrue(e.getMessage().contains("output field list is required"),
e.getMessage());
+ }
+
+ @Test
+ void emptySqlIsRejected() {
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql(" ");
+ meta.setFields(List.of(new SparkField("a", "String")));
+
+ HopException e =
+ assertThrows(
+ HopException.class,
+ () -> run(meta, "sql", new HashMap<>(), List.of(), new
Variables()));
+ assertTrue(e.getMessage().contains("SQL statement is empty"),
e.getMessage());
+ }
+
+ @Test
+ void collidingViewNamesAreReportedRatherThanShadowed() {
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT 1 AS one");
+ meta.setFields(List.of(new SparkField("one", "Integer")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ // Both sanitise to "orders_a"
+ map.put("orders a", ordersDataset());
+ map.put("orders_a", ordersDataset());
+
+ HopException e =
+ assertThrows(
+ HopException.class,
+ () -> run(meta, "sql", map, List.of("orders a", "orders_a"), new
Variables()));
+ assertTrue(e.getMessage().contains("both map to view"), e.getMessage());
+ }
+
+ @Test
+ void invalidSqlReportsTheRegisteredViews() {
+ Dataset<Row> orders = ordersDataset();
+
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setSql("SELECT * FROM does_not_exist");
+ meta.setFields(List.of(new SparkField("a", "String")));
+
+ Map<String, Dataset<Row>> map = new HashMap<>();
+ map.put("orders", orders);
+
+ HopException e =
+ assertThrows(
+ HopException.class, () -> run(meta, "sql", map, List.of("orders"),
new Variables()));
+ assertTrue(e.getMessage().contains("Registered views"), e.getMessage());
+ assertTrue(e.getMessage().contains("orders"), e.getMessage());
+ }
+
+ /** Builds the TransformMeta/PipelineMeta wiring and runs the handler,
returning its Dataset. */
+ private static Dataset<Row> run(
+ SparkSqlMeta meta,
+ String transformName,
+ Map<String, Dataset<Row>> transformDatasetMap,
+ List<String> previousTransformNames,
+ IVariables variables)
+ throws HopException {
+
+ TransformMeta sqlTransform = new TransformMeta(transformName, meta);
+ sqlTransform.setTransformPluginId(SparkConst.SPARK_SQL_PLUGIN_ID);
+
+ PipelineMeta pipelineMeta = new PipelineMeta();
+ pipelineMeta.addTransform(sqlTransform);
+
+ List<TransformMeta> previousTransforms = new ArrayList<>();
+ for (String name : previousTransformNames) {
+ TransformMeta previous = new TransformMeta(name, meta);
+ pipelineMeta.addTransform(previous);
+ previousTransforms.add(previous);
+ }
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("customer_id"));
+ rowMeta.addValueMeta(new ValueMetaInteger("amount"));
+
+ new SparkSqlHandler()
+ .handleTransform(
+ LogChannel.GENERAL,
+ variables,
+ "spark",
+ new SparkPipelineRunConfiguration(),
+ new MemoryMetadataProvider(),
+ "{}",
+ pipelineMeta,
+ sqlTransform,
+ transformDatasetMap,
+ spark,
+ rowMeta,
+ previousTransforms,
+ null);
+
+ return transformDatasetMap.get(transformName);
+ }
+
+ private static Dataset<Row> ordersDataset() {
+ StructType schema =
+ new StructType(
+ new StructField[] {
+ DataTypes.createStructField("customer_id", DataTypes.StringType,
false),
+ DataTypes.createStructField("amount", DataTypes.LongType, false)
+ });
+ List<Row> rows =
+ Arrays.asList(
+ RowFactory.create("c1", 10L),
+ RowFactory.create("c1", 20L),
+ RowFactory.create("c2", 7L),
+ RowFactory.create("c1", 0L));
+ return spark.createDataFrame(rows, schema);
+ }
+
+ private static Dataset<Row> customersDataset() {
+ StructType schema =
+ new StructType(
+ new StructField[] {
+ DataTypes.createStructField("id", DataTypes.StringType, false),
+ DataTypes.createStructField("name", DataTypes.StringType, false)
+ });
+ List<Row> rows =
+ Arrays.asList(RowFactory.create("c1", "alice"),
RowFactory.create("c2", "bob"));
+ return spark.createDataFrame(rows, schema);
+ }
+}
diff --git
a/plugins/engines/spark/src/test/java/org/apache/hop/spark/transforms/sql/SparkSqlMetaTest.java
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/transforms/sql/SparkSqlMetaTest.java
new file mode 100644
index 0000000000..90e98a7ae9
--- /dev/null
+++
b/plugins/engines/spark/src/test/java/org/apache/hop/spark/transforms/sql/SparkSqlMetaTest.java
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 org.apache.hop.spark.transforms.sql;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.List;
+import org.apache.hop.core.HopEnvironment;
+import org.apache.hop.core.row.IRowMeta;
+import org.apache.hop.core.row.IValueMeta;
+import org.apache.hop.core.row.RowMeta;
+import org.apache.hop.core.row.value.ValueMetaString;
+import org.apache.hop.core.variables.Variables;
+import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
+import org.apache.hop.spark.transforms.io.SparkField;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+class SparkSqlMetaTest {
+
+ @BeforeAll
+ static void init() throws Exception {
+ HopEnvironment.init();
+ }
+
+ @Test
+ void defaultViewNameReplacesCharactersThatAreNotSqlSafe() {
+ assertEquals("orders", SparkSqlMeta.defaultViewName("orders"));
+ assertEquals("Read_orders__raw_", SparkSqlMeta.defaultViewName("Read
orders (raw)"));
+ assertEquals("a_b_c", SparkSqlMeta.defaultViewName("a-b.c"));
+ assertEquals("keep_underscores",
SparkSqlMeta.defaultViewName("keep_underscores"));
+ }
+
+ @Test
+ void defaultViewNameEscapesALeadingDigit() {
+ assertEquals("_1_orders", SparkSqlMeta.defaultViewName("1 orders"));
+ assertEquals("_2024", SparkSqlMeta.defaultViewName("2024"));
+ }
+
+ @Test
+ void defaultViewNameHandlesNullAndEmpty() {
+ assertNull(SparkSqlMeta.defaultViewName(null));
+ assertNull(SparkSqlMeta.defaultViewName(""));
+ }
+
+ @Test
+ void findViewNameOverrideIgnoresBlankAndUnknownEntries() {
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setViews(
+ List.of(
+ new SparkSqlView("orders", "o"),
+ new SparkSqlView("customers", ""),
+ new SparkSqlView("blank", null)));
+
+ assertEquals("o", meta.findViewNameOverride("orders"));
+ assertNull(meta.findViewNameOverride("customers"));
+ assertNull(meta.findViewNameOverride("blank"));
+ assertNull(meta.findViewNameOverride("unknown"));
+ assertNull(meta.findViewNameOverride(null));
+ }
+
+ @Test
+ void getFieldsReplacesTheIncomingRowWithTheDeclaredFields() throws Exception
{
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setFields(List.of(new SparkField("name", "String"), new
SparkField("total", "Integer")));
+
+ IRowMeta rowMeta = new RowMeta();
+ rowMeta.addValueMeta(new ValueMetaString("carried_over"));
+
+ meta.getFields(rowMeta, "sql", null, null, new Variables(), new
MemoryMetadataProvider());
+
+ assertEquals(2, rowMeta.size());
+ assertEquals("name", rowMeta.getValueMeta(0).getName());
+ assertEquals(IValueMeta.TYPE_STRING, rowMeta.getValueMeta(0).getType());
+ assertEquals("total", rowMeta.getValueMeta(1).getName());
+ assertEquals(IValueMeta.TYPE_INTEGER, rowMeta.getValueMeta(1).getType());
+ }
+
+ @Test
+ void getFieldsSkipsUnnamedFields() throws Exception {
+ SparkSqlMeta meta = new SparkSqlMeta();
+ meta.setFields(List.of(new SparkField("", "String"), new
SparkField("kept", "String")));
+
+ IRowMeta rowMeta = new RowMeta();
+ meta.getFields(rowMeta, "sql", null, null, new Variables(), new
MemoryMetadataProvider());
+
+ assertEquals(1, rowMeta.size());
+ assertEquals("kept", rowMeta.getValueMeta(0).getName());
+ }
+}
