[flink] Scope scan.parallelism to source operators

[Akash3121]

Purpose

Fixes the behavior reported in issue #7905.

Paimon currently applies scan.parallelism directly to the DataStream produced by PaimonDataStreamScanProvider. However, on Flink 1.19 and later, the Table planner uses ParallelismProvider#getParallelism() to distinguish connector-defined source parallelism from the parallelism of downstream operators.

Because PaimonDataStreamScanProvider did not expose the configured parallelism through this contract, the source parallelism propagated through the forward chain. For example:

parallelism.default = 4
scan.parallelism = 16

Before:
Source (16) --forward--> Calc (16)

Expected:
Source (16) --exchange--> Calc (4)

This change reports the explicitly configured scan parallelism to the Flink planner so that it is scoped to the source operator.

Tests

Added and updated coverage to verify:

• An explicitly configured  scan.parallelism  is returned by the runtime provider.
• The produced source retains the configured parallelism.
• Inferred parallelism is not incorrectly exposed as connector-configured planner metadata.
• Existing inferred source-parallelism behavior is preserved.
• System-table scans expose their configured parallelism.
• Empty row-level-modification scans expose their fixed parallelism of  1 .
• The source transformation uses  scan.parallelism  while the downstream result keeps the environment default parallelism.
• Invalid configured parallelism is rejected by the Flink planner.

Validated with:

mvn -pl paimon-flink/paimon-flink-common -am -Pfast-build -Pflink1
-DwildcardSuites=none -DfailIfNoTests=false
-Dtest=DataTableSourceTest,ReadWriteTableITCase#testSourceParallelism+testInferParallelism test

mvn -pl paimon-flink/paimon-flink-common -am -Pfast-build -Pflink2
-DwildcardSuites=none -DfailIfNoTests=false
-Dtest=DataTableSourceTest,ReadWriteTableITCase#testSourceParallelism+testInferParallelism test

Also validated checked compilation for the Flink 1 and Flink 2 profiles and packaged the Flink 1.16, 1.17, and 1.18 adapters.

Extra info

####Changes

• Make  PaimonDataStreamScanProvider  explicitly implement  ParallelismProvider .
• Add an optional source-parallelism value and return it from  getParallelism() .
• Preserve the existing four-argument constructor for source and binary compatibility.
• Propagate explicitly configured  scan.parallelism  from normal table scans and system-table scans to the runtime provider.
• Report parallelism  1  for the internally fixed empty row-level-modification source.
• Retain the existing producer-side parallelism assignment for Flink 1.16–1.18, whose planners do not consume source parallelism from  ParallelismProvider .
• Keep inferred scan parallelism in the existing runtime path instead of exposing it through  getParallelism() .

####Compatibility

Flink 1.19+ consumes  ParallelismProvider#getParallelism()  for table sources and can isolate the configured source parallelism from downstream operators.

Flink 1.16–1.18 do not consume this metadata for sources. The existing direct  DataStream#setParallelism()  behavior is intentionally retained as a compatibility fallback.

Only an explicitly configured  scan.parallelism  is reported to the planner. Inferred parallelism depends on the runtime  StreamExecutionEnvironment  and split discovery, which are unavailable when the planner queries the provider, so inference behavior remains unchanged.

Changelog

• Bug fix:  scan.parallelism  is now scoped to the Paimon source operator on Flink 1.19 and later instead of propagating to downstream operators.

Notes for reviewers

• Keeping both  getParallelism()  and the existing producer-side  setParallelism()  is intentional. The former fixes planner behavior on Flink 1.19+, while the latter preserves support for older Flink versions.
• Inferred parallelism intentionally returns  Optional.empty()  from the provider because it is calculated later using the runtime environment and scan splits.
• The provider explicitly implements  ParallelismProvider  even though newer  DataStreamScanProvider  versions inherit it. This keeps the method valid against older Flink APIs where  ParallelismProvider  already exists but is not a parent of  DataStreamScanProvider