(flink) branch release-1.19 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 release-1.19
in repository https://gitbox.apache.org/repos/asf/flink.git


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

commit aa277c283795953b6e8717f0e0f4f30341c05d46
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.
    
    Signed-off-by: Jiangjie (Becket) Qin <be...@gmail.com>
---
 .../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 312bd0e0b58..1930e2fd0e0 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
@@ -313,7 +313,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 =>