(flink) branch master updated: [FLINK-37833] [table-runtime] Fix BatchExecExchange codegen for BINARY type by evaluating Unsafe.arrayBaseOffset(byte[].class) in TM rather than in JM (#26592)

[jq...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jqin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/master by this push:
     new 0fcde771ba2 [FLINK-37833] [table-runtime] Fix BatchExecExchange codegen for BINARY type by evaluating Unsafe.arrayBaseOffset(byte[].class) in TM rather than in JM (#26592)
0fcde771ba2 is described below

commit 0fcde771ba23eae7f1756d8961c385c2d7a16b88
Author: Venkata krishnan Sowrirajan <vs...@linkedin.com>
AuthorDate: Wed Jun 11 08:10:50 2025 +0530

    [FLINK-37833] [table-runtime] Fix BatchExecExchange codegen for BINARY type by evaluating Unsafe.arrayBaseOffset(byte[].class) in TM rather than in JM (#26592)
    
    Fix HashPartitioner codegen for BINARY/VARBINARY type by evaluating BYTE_ARRAY_BASE_OFFSET in TM instead of JM.
    
    The issue is, if JM memory is set > 32G while TM memory is set < 32G,
    this causes JVM to treat the JAVA process > 32G as large heap JVM. This
    can impact Unsafe behavior. For eg: UNSAFE.arrayBaseOffset(byte[].class)
    will return 24 for large heap JVM while 16 for others.
    
    Due to this, the tasks that run on TM (<32 G while JM > 32G or vice
    versa) that try to read the byte[] for MurmurHash read wrong memory locations.
---
 .../scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala
index 0f946b173dc..e44a849d1c2 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/codegen/CodeGenUtils.scala
@@ -318,7 +318,11 @@ object CodeGenUtils {
       case BOOLEAN =>
         s"${className[JBoolean]}.hashCode($term)"
       case BINARY | VARBINARY =>
-        s"${className[MurmurHashUtil]}.hashUnsafeBytes($term, $BYTE_ARRAY_BASE_OFFSET, $term.length)"
+        // Instead of computing the BYTE_ARRAY_BASE_OFFSET value in JM, generate the code
+        // and evaluate it in TM. This is required so that byte array offset will be consistent.
+        // See FLINK-37833 for more details.
+        s"${className[MurmurHashUtil]}.hashUnsafeBytes($term," +
+          s" ${className[BinaryRowDataUtil]}.BYTE_ARRAY_BASE_OFFSET, $term.length)"
       case DECIMAL =>
         s"$term.hashCode()"
       case TINYINT =>