[Feature Request]: Native Spark SQL support for the Spark pipeline engine

[mattcasters]

What would you like to happen?

Add first-class Spark SQL support on Hop’s native Spark pipeline engine (see #7486).

Operators should be able to express multi-input joins, filters, windows, projections, and aggregations as SQL compiled by Spark’s Catalyst optimizer against intermediate pipeline Datasets — without chaining Hop Join/Filter/Calculator transforms or pushing SQL through JDBC / mapPartitions.

This builds on the dedicated Spark 4.1.x engine (no Beam). Classic Hop SQL transforms (ExecSql, Table Input, Dynamic SQL row) target RDBMS connections and are the wrong abstraction for Dataset-scoped SQL.

---

Motivation

Gap today	Why Spark SQL helps
Complex multi-table logic needs many Hop transforms	SQL is natural for analytics-style pipelines
Chained Dataset/Hop steps cannot share whole-query planning	A single spark.sql statement gets Catalyst (joins, predicates, codegen)
Classic SQL transforms use JDBC	They do not compile SQL against hop intermediate Datasets
File I/O already established as Spark-only transforms	Same product model (supportedEngines = SparkPipelineEngine)

Reference: Apache Spark SQL — structured data processing via SQL and the Dataset API.

---

Goals

1. Dedicated Spark SQL transform (SparkSql), visible only for the native Spark engine.
2. Multi-input: N enabled incoming hops registered as temp views; SQL can JOIN / UNION them. Zero-input catalog / VALUES SQL allowed when the transform is on the active graph.
3. Driver-side Hop variable substitution (${...}) in SQL text before spark.sql(...) (string concat, not bind parameters).
4. Correct Hop IRowMeta for design-time field resolution and following generic mapPartitions handlers — v1 requires an explicit non-empty output field list.
5. Metrics, logging, and engine compatibility consistent with other native handlers.
6. Works under hop-run, GUI local[*], and spark-submit via MainSpark (no new deploy mode).
7. Unit + integration tests with local SparkSession; README + user-manual docs.

Non-goals (v1)

• Streaming / Structured Streaming SQL (engine is batch-only)
• Making classic JDBC SQL transforms “work natively” without a new transform
• Full Spark catalog governance UI
• Cross-transform registered output views / hop-less catalog composition
• Per-row dynamic SQL from input field values
• Replacing native Merge Join / Memory Group By (SQL is complementary)
• Beam Spark or other engines
• Productized session hardening (UDF blacklist, JDBC datasource sandbox) — use cluster Spark security

---

Proposed design

Product shape

New canvas transform: Spark SQL (plugin id SparkSql).

• Category: Big Data (same as Spark File Input/Output)
• Engine restriction: native Spark only
• Local engine: metadata-only stub (same as Spark File Input)
• Module: plugins/engines/spark (no new Maven module)

Runtime flow

Hop graph (topo-sort)
  → for each predecessor: lookupPreviousDataset → createOrReplaceTempView(alias)
  → resolve variables in SQL on the driver
  → enforce single-statement + default-deny non-SELECT (isQuery heuristic)
  → spark.sql(resolvedSql)
  → project/cast via required SparkField list (TYPED_SOURCE — no string round-trip)
  → SparkNativeMetrics.track(Role.TRANSFORM)
  → put result Dataset in transformDatasetMap
  → drop input temp views after analyze (merge-gated)

Composition is hop-only: temp views exist only for hop predecessors of the current transform. No registerOutputView in v1.

flowchart LR
  O[Spark File Input: orders] --> S[Spark SQL]
  L[Spark File Input: lines] --> S
  S --> OUT[Spark File Output]

Loading

Example SQL (views mapped from hop predecessors):

SELECT o.order_id,
       sum(l.amount) AS total,
       rank() OVER (ORDER BY sum(l.amount) DESC) AS rnk
FROM orders o
JOIN lines l ON o.order_id = l.order_id
GROUP BY o.order_id

Metadata (proposed)

Property	Description
sql	Single Spark SQL query (SELECT / WITH … SELECT / VALUES / TABLE)
viewMappings	Optional hop transform name → SQL view alias (supports ${vars})
fields	Required non-empty output schema (SparkField list) for Hop row meta
allowNonSelect	Default false. Trusted-operator escape hatch for DDL/DML

Default view naming: sanitized predecessor transform name; auto-suffix _2, _3 on collision; fail hard on user-alias clash. Recommend explicit mappings for multi-input SQL.

Multi-input: native handler path (no auto-union). Mandatory use of lookupPreviousDataset for every predecessor so Filter/Switch target streams work (do not copy Merge Join’s direct map get).

Schema / projection

• v1 requires output fields so pipelineMeta.getTransformFields(...) (used by SparkGenericTransformHandler) is never empty.
• Extract dual-mode projection from File Input:
  • STRING_SOURCE — preserve CSV/text File Input cast behavior
  • TYPED_SOURCE — Spark SQL: cast without string intermediate
• Add HopSparkRowConverter.toRowMeta(StructType) for tooling / optional Get Fields (not a v1 empty-fields fallback).

Security

A Spark SQL transform runs with full authority of the job’s SparkSession (catalog, paths, UDFs per cluster config) — same trust level as Spark File Input / any Spark job.

Threat	Mitigation
Untrusted values expanded into SQL via variables	Document: variables are not binds; never inject untrusted strings
DDL/DML / side effects	Default allowNonSelect=false + strengthened isQuery (CTE peel, multi-statement reject, main-verb allowlist)
Secrets in logs	Basic: fingerprint / length; Detailed: full resolved SQL

Observability

• Metrics: existing SparkNativeMetrics with Role.TRANSFORM (same mapPartitions boundary after SQL as other native handlers; Catalyst still optimizes inside the SQL plan)
• Logging: Basic = transform name, effective views, column names; Detailed = full SQL
• Spark UI remains source of truth for SQL stages

---

Key decisions

1. Dedicated SparkSql transform (not classic JDBC SQL)
2. Driver-side spark.sql during graph build (Dataset lineage)
3. Multi-input via temp views; no auto-union
4. Require non-empty output fields in v1 (G4 / generic handler compatibility)
5. Variable substitution is string concat, not parameterization
6. Default reject non-query statements; allowNonSelect escape hatch only
7. SparkFieldProjection: STRING_SOURCE (File I/O) vs TYPED_SOURCE (SQL)
8. Metrics via existing accumulator wrapper
9. Stay in plugins/engines/spark
10. Immediate drop of input temp views after analyze (test-gated)
11. No registerOutputView in v1 — hops only
12. Default alias = sanitized transform name (no magic input)
13. Mandatory lookupPreviousDataset for predecessors
14. Meta + handler + registration in one PR (never fall through to generic mapPartitions)
15. Full SparkSession authority model

---

Implementation plan (PRs)

PR	Scope
PR1	Extract SparkFieldProjection (STRING_SOURCE / TYPED_SOURCE) + StructType → IRowMeta; File I/O stays behavior-preserving
PR2	Spark SQL meta + dialog + handler + converter registration + tests (single merge; no generic fallback)
PR3	README + Antora spark-sql.adoc
PR4 (optional)	Design-time “Get fields” via temporary local SparkSession

Package layout (sketch)

plugins/engines/spark/src/main/java/org/apache/hop/spark/
  transforms/sql/          # SparkSql, Meta, Dialog, ViewMapping
  pipeline/handler/        # SparkSqlHandler, SparkSqlSupport, SparkFieldProjection
  core/HopSparkRowConverter.java   # + toRowMeta
  util/SparkConst.java     # SPARK_SQL_PLUGIN_ID

Test highlights

• Unit: view sanitize/uniqueness, isQuery / multi-statement (incl. CTE-wrapped DML), type matrix, meta check errors
• Integration (local SparkSession): single-input projection, multi-input join, variables, target-stream + multi-input, non-select rejection, metrics, drop-then-count merge gate, zero-input VALUES/SELECT

---

Open questions (non-blocking for v1)

1. Optional GUI “Get fields” via local SparkSession (PR4)
2. Productized CTAS / INSERT sinks later (allowNonSelect is not a productized sink UX)
3. Document cluster spark.sql.caseSensitive behavior
4. Nested Spark types: hard-error non-String Hop fields over nested columns in v1

---

Related

• Parent capability: #7486 — dedicated Spark pipeline execution engine
• Apache Spark SQL
• Native engine module: plugins/engines/spark/

Issue Priority

Priority: 3

Issue Component

Component: Pipelines
Component: Transforms
Component: Documentation