[core] Split large normal files in data evolution compaction

[JingsongLi]

Purpose

Allow Data Evolution tables to resize oversized historical normal files after lowering target-file-size, while preserving the rule that every dedicated BLOB/VECTOR file is fully contained in one normal file's row-ID range.

Add data-evolution.compaction.split-large-files (default: false) and data-evolution.compaction.large-file-ratio (default: 2.0, finite and at least 1.0). An individual normal file strictly exceeding the configured ratio times target-file-size may trigger compaction below compaction.min.file-num. Combined version sizes alone do not bypass the minimum file count.

Before writing, estimate rows per output from the total normal input file size and logical row count, then move cuts inside dedicated files to the ends of their overlapping ranges. Protect ranges across columns and historical versions, plus ranges produced by dedicated compaction in the same batch. Each normal task uses the existing direct writer with automatic size/row rolling disabled and explicitly finishes one file per planned range. Output sizes are approximate and can differ from the target due to compression, data skew, overwritten versions, or dedicated boundaries.

Skip size-only tasks when dedicated ranges prevent any split; ordinary small-file and column-version merging remains available. Dedicated files retain their contents and names unless their separate compaction option requests rewriting them. Preserve row IDs, column sequence metadata, and logical deletions, redistributing deletion vectors to the new normal anchors.

Carry protected ranges in the transient compaction task serializer (version 4; older in-flight/savepoint jobs must restart). Within one Spark COMPACT invocation, bins containing only successfully written normal outputs do not trigger repeated normal rewrites. Include successful conflict-retry outputs; a later COMPACT invocation can select these historical files again.

Validation

Regression coverage includes configurable ratio boundaries, mixed splitting and merging, estimated output ranges, nonzero and extreme row IDs, input-size overflow, cross-column and historical dedicated overlaps, full-range VECTOR protection, same-batch BLOB compaction, task serialization, dedicated-file identity and containment, row and projected reads, column sequence metadata, and deletion vectors. Parquet coverage verifies that its uncompressed write buffer cannot change the planned file boundaries. Buffered-write and write-time row-limit settings cannot cut through a protected range.

Broader validation before the base update passed 481 Java regression cases (457 core and 24 common). On master 06d0653f68, all 84 focused core cases passed again: 45 coordinator/planning, 13 normal-task, and 26 deletion-vector cases.

mvn -pl paimon-core -am -Pspark3 \
  -DwildcardSuites=none -DfailIfNoTests=false \
  '-Dtest=*DataEvolution*Test,CompactCandidateRangeCollectorTest,DataSplitCompatibleTest,ConflictDetectionTest,RollingFileWriterTest,AppendOnlyWriterTest' \
  test

Both focused Spark 3.5 regressions also passed on the updated base: oversized outputs constrained by dedicated ranges are processed once per COMPACT invocation across candidate batches, and the commit observer receives the successful rebased outputs after a same-boundary partial update. Repeating COMPACT starts a new invocation and can process the historical files again.

mvn -pl paimon-spark/paimon-spark-3.5 -am -Pspark3 \
  -DfailIfNoTests=false -DwildcardSuites=none \
  '-Dtest=DataEvolutionCompactCoordinatorTest,DataEvolutionNormalCompactTaskTest,DataEvolutionDeletionVectorTest' \
  '-Dtests=@Paimon Procedure: split oversized files once per compact invocation across batches,@Paimon Procedure: retry rebased compact after same-boundary partial update' \
  -Dtimeout=240 test

Checkstyle, Spotless, and Maven Enforcer passed without fast-build. Local Java validation uses JDK 17 with the existing Byte Buddy agent preloaded and net.bytebuddy.experimental=true for the repository's Mockito version.

The full Spark suite is not claimed as passing: a previous test-lifecycle run hit a ParquetFileReader.open linkage failure because the unshaded reactor paimon-format class shadows Spark's Parquet class. Native Lance and Flink runtime suites were not run locally.

[LsomeYeah]

+1, LGTM！