Repository navigation
Parallel JDBC reads for Table Input on classic and native Spark engines - #9353
rmannibucau wants to merge 2 commits into
Conversation
Add four shared partition settings (jdbc_partition_column, jdbc_lower_bound, jdbc_upper_bound, jdbc_num_partitions) to Table Input metadata and its Options tab, documented in the user manual. TableInputPartitioning mirrors the Spark JDBC data source partition arithmetic so the classic engine (one range per transform copy) and native Spark (one range per partition) read the same rows. Spark: a native SparkJdbcInputHandler uses spark.read.jdbc for the partitioned read and falls back to the generic handler for per-row execution, named parameters, row limits or incoming parameter rows. Stored passwords are decrypted and metrics are forwarded to the fallback handler. The Spark plugin dependency descriptor now exposes the tableinput transform and the shared JDBC drivers. Add guard unit tests and H2 integration tests, and assert the new handler is registered in the converter test.
bamaer
left a comment
There was a problem hiding this comment.
Thanks @rmannibucau, a native partitioned JDBC read is a real win for the Spark engine. The way extra copies read an empty range on the classic engine is neat too.
We built tableinput and engines/spark from this branch on JDK 21 and ran their unit tests and ITs: all green. We also ran a few extra pipelines through HopPipelineMetaToSparkConverter on a local SparkSession against H2.
Blockers
-
Existing Table Inputs on Spark change behaviour, even without partitioning. Registering
TableInputas a native handler (HopPipelineMetaToSparkConverter.java:90,:193) takes every Table Input away from the converter's generic handler:- With more than one main hop into a Table Input, the converter now passes only the first input (
:239). Two parameter sources (SELECT 5,SELECT 200) intoSELECT VAL FROM T WHERE ID = ?give 1 row on this branch and 2 rows on main. - The internal
fallbackhandler never gets the sample-data accumulator or the sampling context.setSampleDataAccumulatorandsetExecutionSamplingContextonly reach the converter's own generic handler, so Table Input no longer produces execution data samples.
Suggested fix: send a Table Input to
SparkJdbcInputHandleronly when partitioning is set, and keep every other Table Input on the converter's generic handler. With that change the two-hop case returns 2 rows again. - With more than one main hop into a Table Input, the converter now passes only the first input (
-
On Spark, a default Table Input never gets the native read, and the fallback then fails.
useNamedParametersistrueby default, soisNativePartitionedRead(SparkJdbcInputHandler.java:90) falls back unless someone unticks that option. The fallback runs a single copy, and the new check atTableInput.java:507stops it:Table Input requires at least 4 copies for the configured JDBC partitions; got 1. The dialog's Preview also runs a single copy, so it hits the same error.Suggested fix: fall back only when the SQL actually contains named parameters (
TableInputSql.parse(sql).hasNamedParameters()), and read the whole result (no slicing) when the transform runs as a single copy. With both changes, a default Table Input reads natively with 4 partitions.
Should fix
- Insert data from transform is not detected. Info hops are not part of
previousTransforms, so theinput != nullcheck (SparkJdbcInputHandler.java:122) misses them and the native read runs the SQL without its parameters:Parameter "#1" is not set. AddingUtils.isEmpty(meta.getLookup())toisNativePartitionedReadwould cover it. - Unrelated
.ycaexclusion. The lines inpom.xml:567and.gitignore:41are for a local tool, not this feature. Could you drop them from this PR?
…ioning for Table Input - Converter: route Table Input to SparkJdbcInputHandler only when a native partitioned read is actually possible (SQL without named/positional placeholders, no lookup, no per-row execution, no incoming rows, no row limit, partitioning configured); otherwise keep the generic handler so multi-input union and sampling wiring are preserved (blocker 1) - isNativePartitionedRead: public 3-arg decision helper; drops the isUseNamedParameters() gate in favor of actual SQL inspection via TableInputSql.parse; parse failure falls back (blocker 2) - Classic TableInput.init(): single copy or preview reads the whole table (partitionSlice null, bounds still validated); error only when copies > 1 and copies < configured (blocker 2) - Lookup (info hop) parameter feeds now fall back to the generic handler instead of failing with 'Parameter apache#1 is not set' (should fix 3) - Shared metadata loader extracted into HopSparkUtil.loadTransformMetadata, reused by SparkBaseTransformHandler and the converter decision (should fix 5) - Remove .yca from .gitignore and RAT excludes (should fix 4) - Tests: rewrite SparkJdbcInputHandlerTest decision cases; new TableInputParametersToSparkIT regression (two parameter sources -> 2 rows); TableInputPartitionedPipelineIT preview/single-copy/insufficient-copies/ invalid-bounds coverage
bamaer
left a comment
There was a problem hiding this comment.
Thanks @rmannibucau for the quick turnaround, the four review items are resolved. We rebuilt d1caaab, ran the module tests and ITs (all green) and re-ran our extra pipelines. Two parameter hops into a Table Input give 2 rows again, and a default Table Input now takes the native read with 4 partitions.
We should have asked for integration tests in our first review and didn't, sorry about that. We're adding them now as a blocker, because this change touches the most-used transform and the Spark engine for every pipeline.
Blocker
- Integration tests. The parts most likely to break are outside what the unit tests and in-JVM ITs can see: the Spark plugin classpath (
lib/jdbcandtableinputindependencies.xml), running from an assembled client, multi-copy.hplpipelines, and a database other than H2. Concretely, we'd need:integration-tests/database: a0008-table-input-partitioned.hplwith 4 copies and the four partition options set, validated against the full row set (no missing or duplicate rows, nulls included). Wire it intomain-0008-table-input.hwfor a new main workflow. The existing0008-table-input-*pipelines show the pattern.integration-tests/spark-native: the same read run throughhop-runon the native Spark engine, validated the same way. That project has no database service yet, so it needs an H2 file database or the Postgres service from the database project.
Should fix
- Pipeline Preview and Debug show every row once per copy.
TableInput.java:511-512reads the whole result whencopies == 1 || preview. Hop GUI's Preview and Debug on the pipeline graph run the real pipeline in preview mode, with its real copy count. A partitioned Table Input with 4 copies then shows each row 4 times, and anything downstream (counts, lookups) is off in the preview.multiCopyPreviewReadsWholeTablePerCopyLikeUnpartitionedasserts exactly that. The dialog Preview already runs a single copy, socopies == 1alone covers it. Dropping|| previewkeeps the slices in a multi-copy preview.
What is the problem being solved?
Table Input in Apache Hop reads in a single database session. Large reads keep one CPU core busy while the storage and network still idle, and on the native Spark engine the transform currently runs only through the generic single-task handler (source transforms are
repartition(1), andHopMapPartitionsFnreplays the source per Spark partition), so a requested parallel read has always had to be written by hand with copy-number predicates.For the impatient
Idea is to five to Apache Hop a bit more of Apache Spark capacity and behavior when using JDBC as input.
Today it is limited by the current code constraint so this PR tries to bring the partitioning feature natively in Apache Hop.
What is being proposed?
Four shared partition settings on Table Input metadata and its Options tab:
jdbc_partition_columnjdbc_lower_boundjdbc_upper_boundjdbc_num_partitionsTableInputPartitioningmirrors the Apache Spark JDBC data-source partition arithmetic exactly (JDBCRelation#columnPartition), so the two engines read the same rows when given the same settings:SparkJdbcInputHandlerreads throughspark.read.jdbcwith Spark partitions; it falls back to the generic handler for per-row execution, named parameters, row limits, or incoming parameter rows. Stored passwords are decrypted and execution metrics are forwarded to the fallback handler.Incomplete or invalid partition settings now fail fast instead of silently falling back. There is no dedicated partition metadata type key; the four properties join the existing transform metadata, and the existing
<exclude>.yca/**</exclude>was not part of this change (see note below).What has been done
@HopMetadataPropertyfields and the new test coverage.TableInputPartitioningand wired classic slicing intoTableInput.SparkJdbcInputHandler, theSparkConst.TABLE_INPUT_PLUGIN_IDregistration, explicit-handler support inHopPipelineMetaToSparkConverter, and the neededdependencies.xmlpaths so the Spark plugin can load the transform and JDBC drivers.SparkJdbcInputHandlerIT), and asserted the new handler registration in the converter test.Verification
mvn -pl plugins/transforms/tableinput,plugins/engines/spark test(unit tests) and the H2 integration tests pass on JDK 21.hop-run.sh -r spark-smokeon the native Spark engine (4 rows read), and confirmed the log shows the native JDBC partitioned read.apache-rat:checkpasses (.yca/**was added to the RAT exclusion list; this was an environment concern unrelated to the feature but is included in the diff but until RAT-578: Support core.excludesFile from global git config creadur-rat#746 is merged there is no real end user workaround for it).Note: this branch was cut from the current
main; the fork'smainis behind, so the PR head is this new branch.