[XiaoHongbo-Hope]

Purpose

Fix conditional self-merge failures when the combined size of Arrow string, binary, or other 32-bit-offset columns exceeds INT32_MAX.

filter_batch previously registered every input RecordBatch in one DataFusion partition. DataFusion could concatenate those batches while evaluating the condition, causing an offset overflow even though each individual Arrow batch was valid.

Changes

• Register each input RecordBatch as a separate DataFusion partition.
• Use collect_partitioned() and rebuild the Arrow table in partition order.
• Disable DataFusion round-robin repartitioning so physical partitions retain the original batch order.
• Add low-memory offset-overflow and real DataFusion multi-batch order regression tests.

The fix preserves the existing Arrow schema instead of casting columns to large_string or large_binary.

Tests

• Real DataFusion 54 order test with four 20,000-row batches
• DataFusion 54 ordering stress check: 30/30 runs ordered after the fix (the old plan reordered 10/10)
• 22 merge-condition unit tests passed
• flake8, py_compile, and git diff --check passed

[JingsongLi]

+1