Parallel JDBC reads for Table Input on classic and native Spark engines

docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/tableinput.adoc

@@ -209,6 +209,17 @@ where id > ${PRM_ID} AND lastdate > ?;
 [#parallel-reading]
 == Parallel reading
 
+=== Built-in partitioned read
+
+Set all four JDBC partition options on the *Options* tab to split a query over a numeric result column.
+The lower and upper bounds determine the partition stride; they do not filter the result, and the first partition includes null values.
+The query must be valid as a subquery and expose the partition column.
+On the classic engine, set the number of transform copies to at least the configured number of partitions; each copy uses a separate connection unless the run configuration shares connections.
+On native Spark, a compatible source uses Spark JDBC partitions; configurations with incoming rows, named parameters, per-row execution or a row limit fall back to the generic handler.
+Concurrent connections may observe different database snapshots, so use an appropriate consistency strategy.
+
+=== Manual partitioning
+
 Keep `NR_COPIES` as a parameter of the pipeline. On the pipeline properties *Parameters* tab, give it a default and a description so the copy count is documented with the pipeline and easy to change for a run. Changing that one value lets you test the limit of the source database's capacity. See xref:fields-parameters-variables.adoc#_parameters[Parameters].
 
 A large read often keeps one database session busy on a single CPU core, while the network and the disks still have capacity. Several copies of the same Table Input share that read. Unless the copies share a connection, each copy runs the query on its own session, and a predicate assigns that copy its own slice of the rows.
@@ -230,7 +241,7 @@ The statements stay the same apart from the copy number. *Limit size* and a SQL
 
 The next transform receives the rows from every copy. One copy of that transform merges the slices into a single stream.
 
-Leave Table Input unpartitioned. A partition schema sets the number of copies itself.
+For this manual approach, leave the JDBC partition options empty. A Hop partition schema sets the number of copies itself.
 
 === Integer keys
 
@@ -401,6 +412,11 @@ On the *Fields* tab you can define the output schema instead of asking the datab
 |Insert data from transform|Optional. Names the hop that should be treated as informational (parameter rows, not the SQL result). Parameter values are read from **all** incoming hops. Leave empty when you only need the connected hops as parameter sources. When the SQL has no bind placeholder, the incoming rows are never bound (only drained when *Execute for each row* is off).
 |Execute for each row?|When incoming hops exist, run the SQL query once for each incoming row. Disable to concatenate all incoming rows into a single parameter list (legacy `WHERE bar IN (?,?,?)` from several one-field rows). Incoming hops with different layouts still produce the mixed-layout warning. When the SQL has no bind placeholder, nothing is ever bound: with this option **off** the incoming rows are only drained and the query runs once without parameters; with it **on** the query runs once per incoming row, still without parameters.
 |Limit size|Maximum number of rows to return from the query. `0` means no limit. See <<limit-size-vs-sql-limit,Limit size vs. SQL LIMIT>> below.
+|JDBC partition column|Numeric column exposed by the query used to split the read.
+|JDBC lower bound|Lower bound for calculating partition strides, not a row filter.
+|JDBC upper bound|Upper bound for calculating partition strides, not a row filter.
+|JDBC number of partitions|Number of partitions; classic execution requires at least this many transform copies.
+
 |Specify output fields|Define the output field list instead of reading column metadata from the database.
 |Validate specified fields|When specifying fields, fail if the query result names or types do not match the list.
 |===

plugins/engines/spark/pom.xml

@@ -226,6 +226,12 @@
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.hop</groupId>
+            <artifactId>hop-transform-tableinput</artifactId>
+            <version>${project.version}</version>
+            <scope>provided</scope>
+        </dependency>
         <dependency>
             <groupId>org.apache.hop</groupId>
             <artifactId>hop-transform-uniquerows</artifactId>
@@ -302,6 +308,14 @@
             <artifactId>h2</artifactId>
             <scope>test</scope>
         </dependency>
+        <!-- H2 database dialect so the JDBC connection metadata used by the partitioned Table
+             Input read resolves its @DatabaseMetaPlugin type in tests. -->
+        <dependency>
+            <groupId>org.apache.hop</groupId>
+            <artifactId>hop-databases-h2</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <!-- Filter Rows for target-stream unit tests -->
         <dependency>
             <groupId>org.apache.hop</groupId>
@@ -373,6 +387,20 @@
                     </execution>
                 </executions>
             </plugin>
+            <!-- Spark ITs (*IT.java) start a local SparkSession against an in-memory H2,
+                 so they run in the failsafe integration-test phase instead of surefire. -->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-failsafe-plugin</artifactId>
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>integration-test</goal>
+                            <goal>verify</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 

plugins/engines/spark/src/main/java/org/apache/hop/spark/core/HopSparkUtil.java

@@ -21,6 +21,7 @@
 import org.apache.hop.core.xml.XmlHandler;
 import org.apache.hop.core.xml.XmlHandlerCache;
 import org.apache.hop.metadata.api.IHopMetadataProvider;
+import org.apache.hop.pipeline.PipelineMeta;
 import org.apache.hop.pipeline.transform.ITransformMeta;
 import org.apache.hop.pipeline.transform.TransformMeta;
 import org.w3c.dom.Document;
@@ -86,4 +87,25 @@ public static void loadTransformMetadataFromXml(
       }
     }
   }
+
+  /**
+   * Loads a detached copy of the transform metadata from the transform's XML snapshot and resolves
+   * its info/target transforms against the pipeline. The live {@link TransformMeta#getTransform()}
+   * instance is never mutated.
+   *
+   * @param meta empty target instance to load into
+   * @param transformMeta the transform whose metadata snapshot is loaded
+   * @param metadataProvider metadata provider used by {@code loadXml}
+   * @param pipelineMeta pipeline the transform belongs to (for info/target resolution)
+   */
+  public static void loadTransformMetadata(
+      ITransformMeta meta,
+      TransformMeta transformMeta,
+      IHopMetadataProvider metadataProvider,
+      PipelineMeta pipelineMeta)
+      throws HopException {
+    loadTransformMetadataFromXml(
+        transformMeta.getName(), meta, transformMeta.getXml(), metadataProvider);
+    meta.searchInfoAndTargetTransforms(pipelineMeta.getTransforms());
+  }
 }

plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/HopPipelineMetaToSparkConverter.java

@@ -38,6 +38,7 @@
 import org.apache.hop.pipeline.config.PipelineRunConfiguration;
 import org.apache.hop.pipeline.transform.BaseTransform;
 import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.pipeline.transforms.tableinput.TableInputMeta;
 import org.apache.hop.spark.core.HopSparkUtil;
 import org.apache.hop.spark.core.SparkExecutionDataAccumulator;
 import org.apache.hop.spark.core.SparkNativeMetricsListener;
@@ -47,6 +48,7 @@
 import org.apache.hop.spark.pipeline.handler.SparkFileInputHandler;
 import org.apache.hop.spark.pipeline.handler.SparkFileOutputHandler;
 import org.apache.hop.spark.pipeline.handler.SparkGenericTransformHandler;
+import org.apache.hop.spark.pipeline.handler.SparkJdbcInputHandler;
 import org.apache.hop.spark.pipeline.handler.SparkLakeTableInputHandler;
 import org.apache.hop.spark.pipeline.handler.SparkLakeTableMaintenanceHandler;
 import org.apache.hop.spark.pipeline.handler.SparkLakeTableMergeHandler;
@@ -85,7 +87,8 @@ public class HopPipelineMetaToSparkConverter {
           SparkConst.SPARK_LAKE_TABLE_OUTPUT_PLUGIN_ID,
           SparkConst.SPARK_LAKE_TABLE_MERGE_PLUGIN_ID,
           SparkConst.SPARK_LAKE_TABLE_MAINTENANCE_PLUGIN_ID,
-          SparkConst.SPARK_SQL_PLUGIN_ID);
+          SparkConst.SPARK_SQL_PLUGIN_ID,
+          SparkConst.TABLE_INPUT_PLUGIN_ID);
 
   /**
    * Plugin ids that must not run as partition-local Hop mini-pipelines. Keep in lockstep with
@@ -188,6 +191,7 @@ protected void addDefaultTransformHandlers() {
     transformHandlers.put(
         SparkConst.SPARK_LAKE_TABLE_MAINTENANCE_PLUGIN_ID, new SparkLakeTableMaintenanceHandler());
     transformHandlers.put(SparkConst.SPARK_SQL_PLUGIN_ID, new SparkSqlHandler());
+    transformHandlers.put(SparkConst.TABLE_INPUT_PLUGIN_ID, new SparkJdbcInputHandler());
   }
 
   public void validatePipeline() throws HopException {
@@ -222,13 +226,30 @@ public Dataset<Row> createDataset(SparkSession spark) throws HopException {
 
       for (TransformMeta transformMeta : transforms) {
         String pluginId = transformMeta.getTransformPluginId();
-        ISparkPipelineTransformHandler handler =
-            transformHandlers.getOrDefault(pluginId, genericTransformHandler);
-        boolean nativeHandler = transformHandlers.containsKey(pluginId);
 
         List<TransformMeta> previousTransforms =
             pipelineMeta.findPreviousTransforms(transformMeta, false);
 
+        // Per-transform handler selection: Table Input only runs through its native JDBC handler
+        // when the partitioned read actually applies; any other configuration (parameters in the
+        // SQL, per-row execution, incoming parameter stream, lookup, row limit) keeps the generic
+        // handler so the multi-input union below and driver-side sampling are preserved.
+        ISparkPipelineTransformHandler handler =
+            transformHandlers.getOrDefault(pluginId, genericTransformHandler);
+        if (SparkConst.TABLE_INPUT_PLUGIN_ID.equals(pluginId)
+            && handler instanceof SparkJdbcInputHandler) {
+          TableInputMeta tableInputMeta = new TableInputMeta();
+          HopSparkUtil.loadTransformMetadata(
+              tableInputMeta, transformMeta, metadataProvider, pipelineMeta);
+          if (SparkJdbcInputHandler.isNativePartitionedRead(
+              tableInputMeta, variables, !previousTransforms.isEmpty())) {
+            handler = transformHandlers.get(pluginId);
+          } else {
+            handler = genericTransformHandler;
+          }
+        }
+        boolean nativeHandler = handler != genericTransformHandler;
+
         Dataset<Row> input = null;
         IRowMeta rowMeta;
         if (previousTransforms.isEmpty()) {

plugins/engines/spark/src/main/java/org/apache/hop/spark/pipeline/handler/SparkBaseTransformHandler.java

@@ -18,18 +18,17 @@
 package org.apache.hop.spark.pipeline.handler;
 
 import org.apache.hop.core.exception.HopException;
-import org.apache.hop.core.xml.XmlHandler;
 import org.apache.hop.metadata.api.IHopMetadataProvider;
 import org.apache.hop.pipeline.PipelineMeta;
 import org.apache.hop.pipeline.transform.ITransformMeta;
 import org.apache.hop.pipeline.transform.TransformMeta;
+import org.apache.hop.spark.core.HopSparkUtil;
 import org.apache.hop.spark.core.SparkNativeMetrics;
 import org.apache.hop.spark.core.SparkNativeMetricsListener;
 import org.apache.hop.spark.core.SparkTransformMetricsAccumulator;
 import org.apache.hop.spark.pipeline.ISparkPipelineTransformHandler;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
-import org.w3c.dom.Node;
 
 public abstract class SparkBaseTransformHandler implements ISparkPipelineTransformHandler {
 
@@ -70,25 +69,19 @@ public boolean isOutput() {
     return false;
   }
 
-  protected Node getTransformXmlNode(TransformMeta transformMeta) throws HopException {
-    String xml = transformMeta.getXml();
-    return XmlHandler.getSubNode(XmlHandler.loadXmlString(xml), TransformMeta.XML_TAG);
-  }
-
+  /**
+   * Loads a detached copy of this transform's metadata from its XML snapshot (never mutating the
+   * live pipeline instance) and resolves info/target transforms against the pipeline.
+   *
+   * @see HopSparkUtil#loadTransformMetadata(ITransformMeta, TransformMeta, IHopMetadataProvider,
+   *     PipelineMeta)
+   */
   protected void loadTransformMetadata(
       ITransformMeta meta,
       TransformMeta transformMeta,
       IHopMetadataProvider metadataProvider,
       PipelineMeta pipelineMeta)
       throws HopException {
-    // Prefer live instance when classloaders match (normal Hop driver)
-    if (meta.getClass().isInstance(transformMeta.getTransform())) {
-      ITransformMeta live = transformMeta.getTransform();
-      // Copy via XML round-trip so we never mutate the live pipeline meta
-      meta.loadXml(getTransformXmlNode(transformMeta), metadataProvider);
-    } else {
-      meta.loadXml(getTransformXmlNode(transformMeta), metadataProvider);
-    }
-    meta.searchInfoAndTargetTransforms(pipelineMeta.getTransforms());
+    HopSparkUtil.loadTransformMetadata(meta, transformMeta, metadataProvider, pipelineMeta);
   }
 }