[spark] Repartition limited lateral vector search input

[shyjsarah]

Purpose

Spark physically executes a GlobalLimit through a single partition. When that limited plan is used as the left input of a lateral vector_search, all query vectors are therefore processed by one task unless users add a repartition manually.

This change:

• detects a GlobalLimit in the lateral vector-search input;
• inserts a shuffle repartition above the limit when no repartition already exists above it;
• preserves an explicit user repartition placed after the limit;
• adds spark.paimon.vector-search.lateral-join.parallelism, defaulting to 16, to control the generated parallelism.

A repartition below the limit does not help because the global limit gathers its result into one partition, so the rule adds a repartition above that shape.

Tests

Added plan tests covering:

• default repartition for a limited input;
• configured parallelism;
• repartition below a global limit;
• preservation of an explicit repartition above the limit.

All four added plan tests pass locally. Apache CI will run the complete Spark test matrix.

[JingsongLi]

Comment 1 — generated documentation is missing

The new VECTOR_SEARCH_LATERAL_JOIN_PARALLELISM option is missing from docs/generated/spark_connector_configuration.html, which causes ConfigOptionsDocsCompletenessITCase to fail in two CI jobs. Please regenerate and commit the option documentation with mvn package -Pgenerate-docs -pl paimon-docs -nsu -DskipTests -am, after installing the changed jars as noted in paimon-docs/README.md.

Comment 2 — broadcast join bypasses the repartition

hasUnrepartitionedGlobalLimit stops at every non-unary node, so a broadcast join above the limited subquery bypasses this optimization even though the streamed side, and therefore the join output, still has one partition. I reproduced this with a limited left side and a broadcast right side: the physical plan contains GlobalLimit -> Exchange SinglePartition -> BroadcastHashJoin, no generated round-robin exchange, and LateralVectorSearchExec receives an RDD with one partition. Could we handle partition-preserving binary shapes such as broadcast joins, or make this decision where the effective child partitioning is available?

Comment 3 — COALESCE is incorrectly considered sufficient

This treats every RepartitionOperation as sufficient, including a non-shuffle COALESCE. A COALESCE(16) above a GlobalLimit cannot increase its single-partition input: the physical plan remains Coalesce 16 -> GlobalLimit -> Exchange SinglePartition, and I verified that LateralVectorSearchExec still receives an RDD with one partition. Please only stop at an operation that can actually restore parallelism, such as a shuffle repartition, or explicitly continue through non-shuffle repartitions.

[shyjsarah]

Addressed the review comments:

• Regenerated docs/generated/spark_connector_configuration.html.
• Moved the repartition decision to an optimizer rule so CTE definitions can be followed.
• Followed the streamed side of broadcast joins when checking for a single-partition GlobalLimit.
• Continued through non-shuffle COALESCE, while preserving an existing shuffle repartition.
• Added regression coverage for CTE, broadcast join, and COALESCE cases.

Validation:

• Spark 3 TableValuedFunctionsTest: 20/20 passed.
• Spark 4 common modules compile successfully with JDK 17.
• Spotless and generated-doc checks passed.

[JingsongLi]

• The rule only recognizes explicit/resolved BROADCAST hints.
• Spark’s normal statistics-based automatic broadcasting is determined during the physical planning phase; since the logical plan contains no hints, the limited streamed side will not be repartitioned, and there may still be only one partition after the physical BroadcastHashJoin.
• Existing tests also only cover /*+ BROADCAST(d) */. It is recommended to add cases without hints that rely on autoBroadcastJoinThreshold, or to handle them at a stage where physical partitioning can be observed.
• The current TableValuedFunctionsTest passes 20 out of 20 locally, indicating that the implementation within the existing test coverage is correct; the issue is insufficient coverage.

[shyjsarah]

Addressed the statistics-based automatic broadcast case in commit 31bf5d0de1.

• Added a physical-plan fallback in LateralVectorSearchExec.requiredChildDistribution, where Spark's selected broadcast build side is available.
• Followed the streamed side of BroadcastHashJoinExec and BroadcastNestedLoopJoinExec through partition-preserving unary operators.
• Restored the configured lateral-search parallelism only when that streamed path still contains an unrepartitioned GlobalLimit.
• Preserved existing streamed-side parallelism for BuildLeft and stopped when a shuffle already exists.
• Added no-hint, statistics-based broadcast tests for both build sides with AQE enabled and disabled.
• Added a version shim for ClusteredDistribution because Spark 3.2 uses the two-argument constructor.

Validation:

• TableValuedFunctionsTest: 22/22 passed.
• Spark 3.2 and 3.3 compile with Scala 2.12 and 2.13.
• Spark 4.0 compiles with JDK 17.
• Verified the Spark 3.2 shaded bytecode calls the two-argument ClusteredDistribution constructor.