[GLUTEN-10546][FLINK] Support all flink operators for nexmark

[shuai-xu]

What changes are proposed in this pull request?

This pr contains:

1. support all the operators use for nexmark test, e.g. GroupAgg, Rank, TopN
2. nexmark q18, q19, q20, q21 run pass, while the result is not equal to flink yet.
3. support checkpoint related interfaces.

How was this patch tested?

This pr is tested by manually run nexmark tests.

[github-actions]

#10546

[philo-he]

Some minor comments. Thanks.

[philo-he]

ditto.

[philo-he]

shiftTimeZone and windowAssigner are not used after the modification. Maybe, we can remove it.

[shuai-xu]

shiftTimeZone is used, so keep it.

[ryyyyyy1]

try to run but it seems to fail due to memory leak error:
E20250830 11:49:14.568943 3949495 MemoryManager.cc:264] [Velox4J MemoryManager DTOR] Memory leak found on Velox memory pool: Memory Pool[Decoding Memory Pool LEAF root[root] parent[root] MALLOC track-usage thread-safe]<unlimited max capacity capacity 128.00MB used 512B available 1023.50KB reservation [used 512B, reserved 1.00MB, min 0B] counters [allocs 4, frees 0, reserves 0, releases 0, collisions 0])>. Please make sure your code released all opened resources already. E20250830 11:49:14.569058 3949495 MemoryManager.cc:264] [Velox4J MemoryManager DTOR] Memory leak found on Velox memory pool: Memory Pool[root AGGREGATE root[root] parent[null] MALLOC track-usage thread-safe]<unlimited max capacity capacity 128.00MB used 512B available 0B reservation [used 0B, reserved 1.00MB, min 0B] counters [allocs 0, frees 0, reserves 0, releases 0, collisions 0])>. Please make sure your code released all opened resources already. E20250830 11:49:14.569092 3949495 MemoryPool.cpp:461] [MEM] Memory leak (Used memory): Memory Pool[Decoding Memory Pool LEAF root[root] parent[root] MALLOC track-usage thread-safe]<unlimited max capacity capacity 128.00MB used 512B available 1023.50KB reservation [used 512B, reserved 1.00MB, min 0B] counters [allocs 4, frees 0, reserves 0, releases 0, collisions 0])> E20250830 11:49:14.569126 3949495 Exceptions.h:66] Line: velox4j/src/main/cpp/main/velox4j/memory/MemoryManager.cc:92, Function:removePool, Expression: pool->reservedBytes() == 0 (1048576 vs. 0), Source: RUNTIME, ErrorCode: INVALID_STATE terminate called after throwing an instance of 'facebook::velox::VeloxRuntimeError' what(): Exception: VeloxRuntimeError Error Source: RUNTIME Error Code: INVALID_STATE Reason: (1048576 vs. 0) Retriable: False Expression: pool->reservedBytes() == 0 Function: removePool File: velox4j/src/main/cpp/main/velox4j/memory/MemoryManager.cc Line: 92 Stack trace: 0 _ZN8facebook5velox7process10StackTraceC1Ei 1 _ZN8facebook5velox14VeloxExceptionC1EPKcmS3_St17basic_string_viewIcSt11char_traitsIcEES7_S7_S7_bNS1_4TypeES7_ 2 _ZN8facebook5velox6detail14veloxCheckFailINS0_17VeloxRuntimeErrorERKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEEEEvRKNS1_18VeloxCheckFailArgsET0_ 3 _ZN7velox4j12_GLOBAL__N_120ListenableArbitrator10removePoolEPN8facebook5velox6memory10MemoryPoolE 4 _ZN8facebook5velox6memory13MemoryManager8dropPoolEPNS1_10MemoryPoolE 5 _ZN8facebook5velox6memory14MemoryPoolImplD2Ev 6 _ZN7velox4j13MemoryManager11tryDestructEv 7 _ZN7velox4j13MemoryManagerD2Ev 8 _ZNSt10_HashtableIjSt4pairIKjSt10shared_ptrIvEESaIS4_ENSt8__detail10_Select1stESt8equal_toIjESt4hashIjENS6_18_Mod_range_hashingENS6_20_Default_ranged_hashENS6_20_Prime_rehash_policyENS6_17_Hashtable_traitsILb0ELb0ELb1EEEE8_M_eraseEmPNS6_15_Hash_node_baseEPNS6_10_Hash_nodeIS4_Lb0EEE 9 _ZN7velox4j11ResourceMapISt10shared_ptrIvEE5eraseEj 10 _ZN7velox4j11ObjectStore15releaseInternalEj 11 _ZN7velox4j12_GLOBAL__N_116releaseCppObjectEP7JNIEnv_P8_jobjectl 12 0x0000ffff80769cfc
any idea to solve?

[shuai-xu]

try to run but it seems to fail due to memory leak error: E20250830 11:49:14.568943 3949495 MemoryManager.cc:264] [Velox4J MemoryManager DTOR] Memory leak found on Velox memory pool: Memory Pool[Decoding Memory Pool LEAF root[root] parent[root] MALLOC track-usage thread-safe]<unlimited max capacity capacity 128.00MB used 512B available 1023.50KB reservation [used 512B, reserved 1.00MB, min 0B] counters [allocs 4, frees 0, reserves 0, releases 0, collisions 0])>. Please make sure your code released all opened resources already. E20250830 11:49:14.569058 3949495 MemoryManager.cc:264] [Velox4J MemoryManager DTOR] Memory leak found on Velox memory pool: Memory Pool[root AGGREGATE root[root] parent[null] MALLOC track-usage thread-safe]<unlimited max capacity capacity 128.00MB used 512B available 0B reservation [used 0B, reserved 1.00MB, min 0B] counters [allocs 0, frees 0, reserves 0, releases 0, collisions 0])>. Please make sure your code released all opened resources already. E20250830 11:49:14.569092 3949495 MemoryPool.cpp:461] [MEM] Memory leak (Used memory): Memory Pool[Decoding Memory Pool LEAF root[root] parent[root] MALLOC track-usage thread-safe]<unlimited max capacity capacity 128.00MB used 512B available 1023.50KB reservation [used 512B, reserved 1.00MB, min 0B] counters [allocs 4, frees 0, reserves 0, releases 0, collisions 0])> E20250830 11:49:14.569126 3949495 Exceptions.h:66] Line: velox4j/src/main/cpp/main/velox4j/memory/MemoryManager.cc:92, Function:removePool, Expression: pool->reservedBytes() == 0 (1048576 vs. 0), Source: RUNTIME, ErrorCode: INVALID_STATE terminate called after throwing an instance of 'facebook::velox::VeloxRuntimeError' what(): Exception: VeloxRuntimeError Error Source: RUNTIME Error Code: INVALID_STATE Reason: (1048576 vs. 0) Retriable: False Expression: pool->reservedBytes() == 0 Function: removePool File: velox4j/src/main/cpp/main/velox4j/memory/MemoryManager.cc Line: 92 Stack trace: 0 _ZN8facebook5velox7process10StackTraceC1Ei 1 _ZN8facebook5velox14VeloxExceptionC1EPKcmS3_St17basic_string_viewIcSt11char_traitsIcEES7_S7_S7_bNS1_4TypeES7_ 2 _ZN8facebook5velox6detail14veloxCheckFailINS0_17VeloxRuntimeErrorERKNSt7__cxx1112basic_stringIcSt11char_traitsIcESaIcEEEEEvRKNS1_18VeloxCheckFailArgsET0_ 3 _ZN7velox4j12_GLOBAL__N_120ListenableArbitrator10removePoolEPN8facebook5velox6memory10MemoryPoolE 4 _ZN8facebook5velox6memory13MemoryManager8dropPoolEPNS1_10MemoryPoolE 5 _ZN8facebook5velox6memory14MemoryPoolImplD2Ev 6 _ZN7velox4j13MemoryManager11tryDestructEv 7 _ZN7velox4j13MemoryManagerD2Ev 8 _ZNSt10_HashtableIjSt4pairIKjSt10shared_ptrIvEESaIS4_ENSt8__detail10_Select1stESt8equal_toIjESt4hashIjENS6_18_Mod_range_hashingENS6_20_Default_ranged_hashENS6_20_Prime_rehash_policyENS6_17_Hashtable_traitsILb0ELb0ELb1EEEE8_M_eraseEmPNS6_15_Hash_node_baseEPNS6_10_Hash_nodeIS4_Lb0EEE 9 _ZN7velox4j11ResourceMapISt10shared_ptrIvEE5eraseEj 10 _ZN7velox4j11ObjectStore15releaseInternalEj 11 _ZN7velox4j12_GLOBAL__N_116releaseCppObjectEP7JNIEnv_P8_jobjectl 12 0x0000ffff80769cfc any idea to solve?

It seems this is not the root reason, the job may fail by some other reason, you can check the log.

[philo-he]

Looks good. Thanks.