mattcasters opened a new issue, #7544: URL: https://github.com/apache/hop/issues/7544
### What would you like to happen? Add first-class **[Spark SQL](https://spark.apache.org/sql/)** support on Hop’s **native Spark pipeline engine** (see [#7486](https://github.com/apache/hop/issues/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 `Dataset`s — 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](https://spark.apache.org/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 ```text 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. ```mermaid flowchart LR O[Spark File Input: orders] --> S[Spark SQL] L[Spark File Input: lines] --> S S --> OUT[Spark File Output] ``` Example SQL (views mapped from hop predecessors): ```sql 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) ```text 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](https://github.com/apache/hop/issues/7486) - [Apache Spark SQL](https://spark.apache.org/sql/) - Native engine module: `plugins/engines/spark/` ### Issue Priority Priority: 3 ### Issue Component Component: Pipelines Component: Transforms Component: Documentation -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
