From 1c818be1154f0bfee4784a723599cb482fce7fd0 Mon Sep 17 00:00:00 2001
From: Nicholas Jiang <programgeek@163.com>
Date: Tue, 4 Aug 2026 23:36:42 +0800
Subject: [PATCH] feat(build): support Linux aarch64 builds

The architecture-specific build logic keyed on PAIMON_CPU_FLAG and
PAIMON_ARMV8_ARCH, neither of which was ever defined, so aarch64 got no
-march= at all. TargetArchitecture.cmake now resolves the target
architecture once, and PAIMON_AARCH64_MARCH selects the -march= value
(default armv8-a; an empty value passes no -march flag). Fixing the
target-dependent behavior this exposed changes one result on x86-64 as
well: out of range float to integer conversion now follows Java.

- Remove dead SIMD build logic: the SSE4.2 CRC32C kernel (Castagnoli,
  never selected, not the persisted zlib CRC-32, so no on-disk value
  changes), the ppc AltiVec probe, and the unread PAIMON_HAVE_NEON /
  ARMV8_CRC / ARMV8_CRYPTO definitions. crc32c_test.cpp now pins the
  checksum to zlib.crc32 reference values.
- Name the package after the host platform, e.g.
  paimon-cpp-linux-aarch64.tar.gz; add --platform and --print-name to
  build_and_package.sh. Lumina is prebuilt for linux-x86_64 only and
  now fails configuration elsewhere with an explicit error.
- Run the same CI matrix on x86_64 and aarch64 (clang-tidy excepted),
  plus a script-tests job for the new CMake-module, packaging-argument
  and asan_symbolize tests; all jobs are required checks.
- Fix char-signedness defects the unsigned-char AArch64 Linux ABI
  exposed: BinaryString UTF-8 leading-byte classification (NumChars,
  Substring and IndexOf counted bytes, not characters), tolower and
  toupper on bytes at or above 0x80, TINYINT min/max aggregation
  (min(-20, 10) returned 10) and VariantValueToString (-20 printed as
  236).
- Replace the undefined float/double to integer static_casts, in the
  cast executor and in the Arrow kernel it called, with
  JavaFloatingToIntegerCast: NaN to 0, out of range saturating at the
  int32/int64 bounds, then narrowing to width. Both the literal and
  the array path use it so stats and data stay consistent.
- Link compiler-rt's builtins into the Clang UBSan build: the 128-bit
  multiplication overflow check calls __muloti4, which libgcc does not
  provide and Clang does not inline on aarch64, so the sanitizer build
  did not link there.
- Pin asan_symbolize.py stdin/stdout to UTF-8 with surrogateescape so
  a byte that is not valid UTF-8 no longer truncates the test log.
- docs: a supported platform matrix, rules on char signedness and on
  float to integer conversion, and Java-consistent results in the type
  change support matrix.
---
 .asf.yaml                                     |  26 +-
 .github/workflows/build_and_test.yaml         |  52 +++-
 .github/workflows/release_candidate.yaml      |   7 +-
 README.md                                     |   3 +-
 build_and_package.sh                          |  51 ++++
 build_support/asan_symbolize.py               |  14 +-
 ci/scripts/build_paimon.sh                    |   6 +
 ci/scripts/test_asan_symbolize.sh             |  95 ++++++
 ci/scripts/test_cmake_modules.sh              |  34 +++
 ci/scripts/test_packaging_args.sh             | 164 ++++++++++
 cmake_modules/DefineOptions.cmake             |  18 ++
 cmake_modules/SetupCxxFlags.cmake             |  84 +++---
 cmake_modules/TargetArchitecture.cmake        |  50 ++++
 cmake_modules/ThirdpartyToolchain.cmake       |  26 +-
 cmake_modules/san-config.cmake                |  22 ++
 .../tests/aarch64_march_option_test.cmake     | 115 +++++++
 .../tests/target_architecture_test.cmake      | 114 +++++++
 docs/code-style.md                            |  20 +-
 docs/source/building.rst                      |  60 +++-
 docs/source/user_guide/read.rst               |  16 +-
 src/paimon/common/data/binary_string.cpp      |  27 +-
 src/paimon/common/data/binary_string_test.cpp |  20 +-
 src/paimon/common/data/data_define.h          |   4 +
 src/paimon/common/data/data_define_test.cpp   |  11 +
 .../binary_row_partition_computer_test.cpp    |   8 +-
 src/paimon/common/utils/crc32c.cpp            |  41 +--
 src/paimon/common/utils/crc32c.h              |  29 +-
 src/paimon/common/utils/crc32c_test.cpp       | 102 ++++++-
 .../utils/data_converter_utils_test.cpp       |   9 +-
 .../core/casting/cast_executor_test.cpp       | 283 ++++++++++++------
 .../numeric_primitive_cast_executor.cpp       |  70 +++++
 .../casting/numeric_primitive_cast_executor.h |  63 +++-
 .../compact/aggregate/field_max_agg.h         |  12 +
 .../compact/aggregate/field_min_agg.h         |  12 +
 .../aggregate/field_min_max_agg_test.cpp      |   3 +
 35 files changed, 1421 insertions(+), 250 deletions(-)
 create mode 100755 ci/scripts/test_asan_symbolize.sh
 create mode 100755 ci/scripts/test_cmake_modules.sh
 create mode 100755 ci/scripts/test_packaging_args.sh
 create mode 100644 cmake_modules/TargetArchitecture.cmake
 create mode 100644 cmake_modules/tests/aarch64_march_option_test.cmake
 create mode 100644 cmake_modules/tests/target_architecture_test.cmake

diff --git a/.asf.yaml b/.asf.yaml
index 33a74a179..0cc39ed40 100644
--- a/.asf.yaml
+++ b/.asf.yaml
@@ -59,17 +59,31 @@ github:
           app_slug: -1
         - name: "rat-license-check"
           app_slug: -1
-        - name: "asan-ubsan"
+        - name: "script-tests"
           app_slug: -1
-        - name: "tsan"
+        - name: "asan-ubsan-x86_64"
           app_slug: -1
-        - name: "clang-debug"
+        - name: "tsan-x86_64"
           app_slug: -1
-        - name: "clang-release"
+        - name: "clang-debug-x86_64"
           app_slug: -1
-        - name: "gcc-debug"
+        - name: "clang-release-x86_64"
           app_slug: -1
-        - name: "gcc-release"
+        - name: "gcc-debug-x86_64"
+          app_slug: -1
+        - name: "gcc-release-x86_64"
+          app_slug: -1
+        - name: "gcc-debug-aarch64"
+          app_slug: -1
+        - name: "gcc-release-aarch64"
+          app_slug: -1
+        - name: "clang-debug-aarch64"
+          app_slug: -1
+        - name: "asan-ubsan-aarch64"
+          app_slug: -1
+        - name: "clang-release-aarch64"
+          app_slug: -1
+        - name: "tsan-aarch64"
           app_slug: -1
         - name: "gcc8-test"
           app_slug: -1
diff --git a/.github/workflows/build_and_test.yaml b/.github/workflows/build_and_test.yaml
index ab3f22bb0..b19e49069 100644
--- a/.github/workflows/build_and_test.yaml
+++ b/.github/workflows/build_and_test.yaml
@@ -33,31 +33,69 @@ permissions:
   contents: read
 
 jobs:
+  script-tests:
+    name: script-tests
+    runs-on: ubuntu-24.04
+    timeout-minutes: 10
+    steps:
+      - name: Checkout paimon-cpp
+        uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
+      - name: Run CMake module tests
+        shell: bash
+        run: ci/scripts/test_cmake_modules.sh
+      - name: Run packaging argument tests
+        shell: bash
+        run: ci/scripts/test_packaging_args.sh
+      - name: Run asan_symbolize tests
+        shell: bash
+        run: ci/scripts/test_asan_symbolize.sh
+
   build-and-test:
     name: ${{ matrix.name }}
-    runs-on: ubuntu-24.04
+    runs-on: ${{ matrix.runner || 'ubuntu-24.04' }}
     timeout-minutes: 120
     strategy:
       fail-fast: false
       matrix:
         include:
-          - name: gcc-release
+          - name: gcc-release-x86_64
             cc: gcc-14
             cxx: g++-14
             build_args: --build_type Release
-          - name: clang-release
+          - name: clang-release-x86_64
             build_args: --build_type Release
-          - name: gcc-debug
+          - name: gcc-debug-x86_64
             cc: gcc-14
             cxx: g++-14
-          - name: clang-debug
+          - name: clang-debug-x86_64
             fetch_depth: '0' # fetch the PR target branch history for clang-tidy
             build_args: >-
               --check_clang_tidy
               --lint_git_target_commit "origin/${{ github.base_ref || github.event.repository.default_branch }}"
-          - name: asan-ubsan
+          - name: asan-ubsan-x86_64
             build_args: --enable_asan --enable_ubsan
-          - name: tsan
+          - name: tsan-x86_64
+            skip_rust: true
+            build_args: --enable_tsan
+          - name: gcc-debug-aarch64
+            runner: ubuntu-24.04-arm
+            cc: gcc-14
+            cxx: g++-14
+          - name: gcc-release-aarch64
+            runner: ubuntu-24.04-arm
+            cc: gcc-14
+            cxx: g++-14
+            build_args: --build_type Release
+          - name: clang-debug-aarch64
+            runner: ubuntu-24.04-arm
+          - name: asan-ubsan-aarch64
+            runner: ubuntu-24.04-arm
+            build_args: --enable_asan --enable_ubsan
+          - name: clang-release-aarch64
+            runner: ubuntu-24.04-arm
+            build_args: --build_type Release
+          - name: tsan-aarch64
+            runner: ubuntu-24.04-arm
             skip_rust: true
             build_args: --enable_tsan
     steps:
diff --git a/.github/workflows/release_candidate.yaml b/.github/workflows/release_candidate.yaml
index 859b25780..6b5b1ebda 100644
--- a/.github/workflows/release_candidate.yaml
+++ b/.github/workflows/release_candidate.yaml
@@ -99,7 +99,7 @@ jobs:
     name: Verify source archive (${{ matrix.compiler }})
     if: github.ref_type == 'tag'
     needs: archive
-    runs-on: ubuntu-24.04
+    runs-on: ${{ matrix.runner || 'ubuntu-24.04' }}
     timeout-minutes: 180
     strategy:
       fail-fast: false
@@ -107,6 +107,7 @@ jobs:
         compiler:
           - gcc-14
           - clang
+          - gcc-14-aarch64
         include:
           - compiler: gcc-14
             cc: gcc-14
@@ -114,6 +115,10 @@ jobs:
           - compiler: clang
             cc: clang
             cxx: clang++
+          - compiler: gcc-14-aarch64
+            cc: gcc-14
+            cxx: g++-14
+            runner: ubuntu-24.04-arm
     steps:
       - name: Checkout source
         uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v6.0.1
diff --git a/README.md b/README.md
index a912e067d..9161e6f89 100644
--- a/README.md
+++ b/README.md
@@ -49,7 +49,8 @@ Paimon C++ currently provides:
 > Paimon C++ therefore currently treats the `bitmap` global index type as unsupported. The legacy
 > implementation remains in the codebase pending migration to the Java-compatible format.
 
-Note: Only Linux x86_64 builds are currently supported and verified.
+Note: Linux `x86_64` and `aarch64` builds are supported and verified in CI. See the supported
+platform matrix in [docs/source/building.rst](docs/source/building.rst) for other platforms.
 
 ## Building
 
diff --git a/build_and_package.sh b/build_and_package.sh
index f9e671e79..652eaecaa 100755
--- a/build_and_package.sh
+++ b/build_and_package.sh
@@ -26,6 +26,8 @@ MAKE_CLEAN=false
 PACKAGE=false
 CMAKE_OPTIONS=()
 JOBS=""
+PACKAGE_PLATFORM=""
+PRINT_NAME=false
 
 show_help() {
     cat << EOF
@@ -37,6 +39,12 @@ Options:
   -c, --clean       Clean build directory before building
   -p, --package     Package creation
   -j, --jobs <num>  Number of parallel jobs for building (default: auto-detect)
+      --platform <label>
+                    Platform label used in the package name (default: the host,
+                    e.g. linux-aarch64). This only labels the artifact; it does
+                    not configure a cross build. Must match
+                    [A-Za-z0-9][A-Za-z0-9._-]*, i.e. a single path component.
+      --print-name  Print the resolved package name and exit
   -h, --help        Show this help message
 
 CMake Options:
@@ -46,6 +54,7 @@ CMake Options:
 Examples:
   $0 -r -p -j 8 -DPAIMON_BUILD_SHARED=ON -DPAIMON_BUILD_STATIC=OFF
   $0 --debug --clean --package --jobs 4
+  $0 -r -p --platform linux-aarch64
 
 EOF
 }
@@ -82,6 +91,29 @@ while [[ $# -gt 0 ]]; do
                 exit 1
             fi
             ;;
+        --platform=*)
+            # Normalize the equals form onto the two-argument form below, so the
+            # value goes through the same validation.
+            set -- --platform "${1#*=}" "${@:2}"
+            ;;
+        --platform)
+            shift
+            # The label becomes a path component of both the install prefix and
+            # the tarball name, so it must stay a single safe component: the
+            # pattern admits no separator and no shell metacharacter. Requiring a
+            # leading letter or digit additionally stops a following option from
+            # being swallowed as the value.
+            if [[ $# -eq 0 || ! $1 =~ ^[A-Za-z0-9][A-Za-z0-9._-]*$ ]]; then
+                echo "Error: --platform requires a label matching [A-Za-z0-9][A-Za-z0-9._-]*" >&2
+                exit 1
+            fi
+            PACKAGE_PLATFORM="$1"
+            shift
+            ;;
+        --print-name)
+            PRINT_NAME=true
+            shift
+            ;;
         -h|--help)
             show_help
             exit 0
@@ -94,8 +126,27 @@ while [[ $# -gt 0 ]]; do
     esac
 done
 
+# Defaults to the host. Note this only names the artifact: the bundled third-party
+# builds receive just CMAKE_C_COMPILER / CMAKE_CXX_COMPILER (see
+# EP_COMMON_TOOLCHAIN in cmake_modules/ThirdpartyToolchain.cmake) and no toolchain
+# file, sysroot or find-root, so a cross build is not wired up end to end.
+if [ -z "$PACKAGE_PLATFORM" ]; then
+    HOST_OS=$(uname -s | tr '[:upper:]' '[:lower:]')
+    if [ "$HOST_OS" = "darwin" ]; then
+        HOST_OS="macos"
+    fi
+    PACKAGE_PLATFORM="$HOST_OS-$(uname -m)"
+fi
+BUILD_NAME="$BUILD_NAME-$PACKAGE_PLATFORM"
+
+if [ "$PRINT_NAME" = true ]; then
+    echo "$BUILD_NAME"
+    exit 0
+fi
+
 echo "========== Build Configuration =========="
 echo "Build Type: $BUILD_TYPE"
+echo "Package Platform: $PACKAGE_PLATFORM"
 echo "Package Name: $BUILD_NAME"
 echo "Clean Build: $MAKE_CLEAN"
 echo "Package: $PACKAGE"
diff --git a/build_support/asan_symbolize.py b/build_support/asan_symbolize.py
index 854090ae5..a119d9882 100755
--- a/build_support/asan_symbolize.py
+++ b/build_support/asan_symbolize.py
@@ -8,6 +8,7 @@
 #
 #===------------------------------------------------------------------------===#
 import bisect
+import io
 import os
 import re
 import subprocess
@@ -333,8 +334,17 @@ def process_stdin(self):
     if sys.version_info[0] == 2:
       sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', 0)
     else:
-      # Unbuffered output is not supported in Python 3
-      sys.stdout = os.fdopen(sys.stdout.fileno(), 'w')
+      # Test output is not guaranteed to be valid UTF-8, so pin both streams to the same
+      # encoding and round such bytes through surrogateescape, which passes them out
+      # unchanged instead of raising and truncating the rest of the test log. Both the
+      # encoding and the error handler have to be set on both streams: taking either from
+      # the locale or from PYTHONIOENCODING would let them disagree and re-encode the bytes.
+      # The buffers are rewrapped rather than reconfigured, which needs Python 3.7, and
+      # line buffering stands in for the unbuffered output Python 3 does not support.
+      sys.stdin = io.TextIOWrapper(sys.stdin.buffer, encoding='utf-8',
+                                   errors='surrogateescape')
+      sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8',
+                                    errors='surrogateescape', line_buffering=True)
 
     while True:
       line = sys.stdin.readline()
diff --git a/ci/scripts/build_paimon.sh b/ci/scripts/build_paimon.sh
index c55a8a6f5..50983d4da 100755
--- a/ci/scripts/build_paimon.sh
+++ b/ci/scripts/build_paimon.sh
@@ -136,6 +136,12 @@ fi
 if [[ "${enable_tsan}" == "true" ]]; then
     ENABLE_TANTIVY="OFF" # Tantivy's Rust library is not TSAN-instrumented.
 fi
+# CI always builds natively, so the host architecture is the target architecture.
+host_arch=$(uname -m)
+if [[ "${host_arch}" != "x86_64" ]]; then
+    ENABLE_LUMINA="OFF"
+    echo "=== Lumina disabled: no prebuilt artifacts for ${host_arch} ==="
+fi
 
 CMAKE_ARGS=(
     "-G Ninja"
diff --git a/ci/scripts/test_asan_symbolize.sh b/ci/scripts/test_asan_symbolize.sh
new file mode 100755
index 000000000..0859559b4
--- /dev/null
+++ b/ci/scripts/test_asan_symbolize.sh
@@ -0,0 +1,95 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+# Tests that build_support/asan_symbolize.py, which every test binary pipes its
+# output through, passes bytes that are not valid UTF-8 straight through. A test
+# that fails while printing such a byte used to kill the symbolizer, and with it
+# the rest of the test log.
+
+set -euo pipefail
+
+source_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)
+symbolizer="${source_dir}/build_support/asan_symbolize.py"
+python=${PYTHON:-python3}
+# The vendored script emits SyntaxWarnings for its own regexes; they are unrelated noise here.
+export PYTHONWARNINGS=ignore
+
+status=0
+
+# Returns non-zero on failure as well as recording it, so that a caller running it in a
+# subshell, where the recorded status would not propagate, can still see the result.
+check() {
+    local name=$1 input=$2 expected=$3 actual_hex expected_hex result=0
+    echo "=== ${name} ==="
+    # Compare hex dumps so that a mismatch is readable and the shell does not
+    # mangle the bytes on the way. A failing symbolizer is a failed check, not a
+    # reason to abort the script, so the remaining checks still run.
+    if ! actual_hex=$(printf '%b' "${input}" | "${python}" "${symbolizer}" | od -An -tx1 |
+        tr -d ' \n'); then
+        echo "symbolizer exited non-zero"
+        status=1
+        result=1
+    fi
+    expected_hex=$(printf '%b' "${expected}" | od -An -tx1 | tr -d ' \n')
+    if [[ "${actual_hex}" != "${expected_hex}" ]]; then
+        echo "expected: ${expected_hex}"
+        echo "actual:   ${actual_hex}"
+        status=1
+        result=1
+    fi
+    return "${result}"
+}
+
+# A lone 0xff is what a failing TINYINT literal assertion prints. Every byte,
+# valid UTF-8 or not, has to come out unchanged.
+# `|| status=1` keeps `set -e` from aborting on the first failed check.
+check "invalid utf-8 round trips" 'ok\n\xff\xfe binary\nplain\n' \
+    'ok\n\xff\xfe binary\nplain\n' || status=1
+
+# Valid multi byte UTF-8 must not be damaged either: test names contain it.
+check "utf-8 round trips" '\xe4\xb8\xad\xe6\x96\x87\n' '\xe4\xb8\xad\xe6\x96\x87\n' || status=1
+
+# The bytes have to survive whatever the environment asks Python to use: an encoding taken
+# from PYTHONIOENCODING on one stream and from the locale on the other would re-encode them.
+(
+    export PYTHONIOENCODING=latin-1
+    check "invalid utf-8 round trips under PYTHONIOENCODING" 'ok\n\xff\xfe\n' \
+        'ok\n\xff\xfe\n' || status=1
+    # status is set in this subshell and does not reach the caller, so report it as the exit
+    # code, which stays correct if another check is added here.
+    exit "${status}"
+) || status=1
+
+# An invalid byte must not stop the lines that follow it from being processed:
+# the stack frame below is still rewritten by the symbolizer. Its stderr is dropped
+# because addr2line reports the fake binary path there even when the check passes.
+echo "=== keeps processing after an invalid byte ==="
+if ! output=$(printf '%b' '\xff\n    #0 0x7f6e35cf2e45  (/blah/foo.so+0x11fe45)\ntail\n' |
+    "${python}" "${symbolizer}" 2>/dev/null); then
+    echo "symbolizer exited non-zero"
+    status=1
+fi
+# "#0" and "tail" alone would also match unprocessed input, so assert that the frame was
+# actually rewritten: the symbolized form gains " in" and loses the raw binary path.
+if [[ "${output}" != *"#0"* || "${output}" != *"tail"* || "${output}" != *" in"* ||
+    "${output}" == *"/blah/foo.so"* ]]; then
+    echo "frame was not rewritten after the invalid byte, got:"
+    echo "${output}"
+    status=1
+fi
+
+exit "${status}"
diff --git a/ci/scripts/test_cmake_modules.sh b/ci/scripts/test_cmake_modules.sh
new file mode 100755
index 000000000..31acb6737
--- /dev/null
+++ b/ci/scripts/test_cmake_modules.sh
@@ -0,0 +1,34 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Runs the CMake module tests under cmake_modules/tests. These are script-mode
+# tests: they need neither a toolchain nor a configured build tree, so they stay
+# separate from ctest and run in seconds.
+
+set -euo pipefail
+
+source_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)
+
+status=0
+for test_script in "${source_dir}"/cmake_modules/tests/*_test.cmake; do
+    echo "=== ${test_script##*/} ==="
+    if ! cmake -P "${test_script}"; then
+        status=1
+    fi
+done
+
+exit "${status}"
diff --git a/ci/scripts/test_packaging_args.sh b/ci/scripts/test_packaging_args.sh
new file mode 100755
index 000000000..5c9fb326b
--- /dev/null
+++ b/ci/scripts/test_packaging_args.sh
@@ -0,0 +1,164 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# Argument-handling tests for build_and_package.sh. They only exercise
+# --print-name, which resolves the package name and exits before any build step,
+# so no toolchain is needed.
+
+# No `set -e`: every check runs the script under test and inspects its exit
+# status, so a non-zero status is data here, not a reason to abort.
+set -uo pipefail
+
+source_dir=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)
+script="${source_dir}/build_and_package.sh"
+
+checks=0
+failures=0
+
+# A private directory: a fixed /tmp path would collide with a concurrent run and
+# could make the script delete a file it did not create.
+work_dir=$(mktemp -d)
+trap 'rm -rf "${work_dir}"' EXIT
+sentinel="${work_dir}/must-not-exist"
+
+# --print-name is passed FIRST, never last: the arguments under test have to keep
+# their own position, otherwise a trailing --print-name would become the value of
+# a preceding --platform and the argument-exhaustion branch could never be
+# reached. Parsing is order independent, so leading it changes nothing else.
+expect_name() {
+    local description=$1 expected=$2
+    shift 2
+    local actual
+    checks=$((checks + 1))
+    if ! actual=$("${script}" --print-name "$@" 2>&1); then
+        failures=$((failures + 1))
+        echo "FAIL ${description}: exited non-zero: ${actual}"
+    elif [[ "${actual}" != "${expected}" ]]; then
+        failures=$((failures + 1))
+        echo "FAIL ${description}: expected '${expected}', got '${actual}'"
+    fi
+}
+
+# Asserting the diagnostic, not just a non-zero exit: the script runs under
+# `set -u`, so a missing validation branch would still abort -- with an "unbound
+# variable" crash instead of a usable message.
+expect_rejected() {
+    local description=$1
+    shift
+    local output status
+    checks=$((checks + 1))
+    output=$("${script}" --print-name "$@" 2>&1)
+    status=$?
+    if [[ "${status}" -eq 0 ]]; then
+        failures=$((failures + 1))
+        echo "FAIL ${description}: expected a non-zero exit, but the value was accepted"
+    elif [[ "${output}" != *"--platform requires a label"* ]]; then
+        failures=$((failures + 1))
+        echo "FAIL ${description}: expected the --platform diagnostic, got: ${output}"
+    fi
+}
+
+# The default label is derived from `uname`, so stub it on PATH to check the
+# derivation for hosts this machine is not, including the Darwin -> macos mapping.
+expect_host_default() {
+    local description=$1 uname_s=$2 uname_m=$3 expected=$4
+    shift 4
+    local stub_dir actual status
+    checks=$((checks + 1))
+    stub_dir=$(mktemp -d "${work_dir}/stub.XXXXXX")
+    cat > "${stub_dir}/uname" <<EOF
+#!/bin/sh
+case "\$1" in
+    -s) echo "${uname_s}" ;;
+    -m) echo "${uname_m}" ;;
+    *) echo "unexpected uname argument: \$1" >&2; exit 1 ;;
+esac
+EOF
+    chmod +x "${stub_dir}/uname"
+    actual=$(PATH="${stub_dir}:${PATH}" "${script}" --print-name "$@" 2>&1)
+    status=$?
+    if [[ "${status}" -ne 0 ]]; then
+        failures=$((failures + 1))
+        echo "FAIL ${description}: exited ${status}: ${actual}"
+    elif [[ "${actual}" != "${expected}" ]]; then
+        failures=$((failures + 1))
+        echo "FAIL ${description}: expected '${expected}', got '${actual}'"
+    fi
+}
+
+host_platform="$(uname -s | tr '[:upper:]' '[:lower:]')"
+if [[ "${host_platform}" == "darwin" ]]; then
+    host_platform="macos"
+fi
+host_platform="${host_platform}-$(uname -m)"
+
+expect_name "release defaults to the host platform" "paimon-cpp-${host_platform}"
+expect_name "debug keeps its own prefix" "paimon-cpp-debug-${host_platform}" --debug
+expect_name "release is the default build type" "paimon-cpp-${host_platform}" --release
+expect_name "--platform overrides the host" "paimon-cpp-linux-aarch64" --platform linux-aarch64
+expect_name "--platform applies to debug too" "paimon-cpp-debug-macos-arm64" -d --platform \
+    macos-arm64
+expect_name "the last --platform wins" "paimon-cpp-linux-aarch64" --platform linux-x86_64 \
+    --platform linux-aarch64
+
+# The documented pattern admits dots, underscores and hyphens; keep that a contract.
+expect_name "a dotted label is accepted" "paimon-cpp-linux.arm64" --platform linux.arm64
+expect_name "an underscored label is accepted" "paimon-cpp-linux_musl-aarch64" --platform \
+    linux_musl-aarch64
+
+# The equals form goes through the same validation as the two-argument form.
+expect_name "the equals form is accepted" "paimon-cpp-linux-aarch64" --platform=linux-aarch64
+expect_rejected "empty equals form" --platform=
+expect_rejected "equals form with a nested path" --platform=linux/x86_64
+
+# Default labels for hosts other than this one.
+expect_host_default "linux aarch64 host" Linux aarch64 "paimon-cpp-linux-aarch64"
+expect_host_default "darwin arm64 host maps to macos" Darwin arm64 "paimon-cpp-macos-arm64"
+expect_host_default "darwin x86_64 host maps to macos" Darwin x86_64 "paimon-cpp-macos-x86_64"
+expect_host_default "debug on a linux aarch64 host" Linux aarch64 \
+    "paimon-cpp-debug-linux-aarch64" --debug
+expect_host_default "--platform still wins over the host" Linux aarch64 \
+    "paimon-cpp-linux-x86_64" --platform linux-x86_64
+
+# --platform as the final argument must be reported as a missing value rather than
+# consuming whatever follows.
+expect_rejected "--platform without a value" --platform
+
+# Nothing that turns the package name into a path, or that a shell would treat as
+# anything but a literal, may be accepted.
+expect_rejected "--platform followed by an option" --platform --debug
+expect_rejected "bare parent directory" --platform ".."
+expect_rejected "parent directory traversal" --platform "../evil"
+expect_rejected "absolute path" --platform "/etc/passwd"
+expect_rejected "nested path" --platform "linux/x86_64"
+expect_rejected "leading dash" --platform "-linux"
+expect_rejected "leading dot" --platform ".linux"
+expect_rejected "empty label" --platform ""
+expect_rejected "command separator" --platform "linux;touch ${sentinel}"
+expect_rejected "command substitution" --platform 'linux$(touch '"${sentinel}"')'
+expect_rejected "whitespace" --platform "linux x86_64"
+
+if [[ -e "${sentinel}" ]]; then
+    failures=$((failures + 1))
+    echo "FAIL a rejected label was still evaluated by a shell"
+fi
+
+if [[ "${failures}" -gt 0 ]]; then
+    echo "${failures} of ${checks} packaging argument checks failed"
+    exit 1
+fi
+echo "All ${checks} packaging argument checks passed"
diff --git a/cmake_modules/DefineOptions.cmake b/cmake_modules/DefineOptions.cmake
index 37b5d4453..21f653155 100644
--- a/cmake_modules/DefineOptions.cmake
+++ b/cmake_modules/DefineOptions.cmake
@@ -18,6 +18,10 @@
 # Borrowed the file from Apache Arrow:
 # https://github.com/apache/arrow/blob/apache-arrow-17.0.0/cpp/cmake_modules/DefineOptions.cmake
 
+# Single source of the Arm64 default, applied both to the option below and to the
+# fallback after it.
+set(PAIMON_AARCH64_MARCH_DEFAULT "armv8-a")
+
 macro(set_option_category name)
     set(PAIMON_OPTION_CATEGORY ${name})
     list(APPEND "PAIMON_OPTION_CATEGORIES" ${name})
@@ -102,6 +106,10 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}")
                   "Use Apple SDK libc++ headers when building with upstream Clang on macOS"
                   ON)
 
+    define_option_string(PAIMON_AARCH64_MARCH
+                         "AArch64 value passed to -march=; empty passes no -march flag"
+                         "${PAIMON_AARCH64_MARCH_DEFAULT}")
+
     #----------------------------------------------------------------------
     set_option_category("Test")
 
@@ -262,6 +270,16 @@ if("${CMAKE_SOURCE_DIR}" STREQUAL "${CMAKE_CURRENT_SOURCE_DIR}")
                          SYSTEM)
 endif()
 
+# The option above exists only for a top-level build, but SetupCxxFlags.cmake
+# uses PAIMON_AARCH64_MARCH on AArch64 targets, so an add_subdirectory() consumer
+# needs the same default. Only an undefined value falls back: an explicitly empty
+# value is the opt-out that SetupCxxFlags.cmake answers with no -march flag at
+# all, and any other value, OFF included, is taken literally so a typo fails the
+# compiler probe instead of being coerced silently.
+if(NOT DEFINED PAIMON_AARCH64_MARCH)
+    set(PAIMON_AARCH64_MARCH "${PAIMON_AARCH64_MARCH_DEFAULT}")
+endif()
+
 macro(validate_config)
     foreach(category ${PAIMON_OPTION_CATEGORIES})
         set(option_names ${PAIMON_${category}_OPTION_NAMES})
diff --git a/cmake_modules/SetupCxxFlags.cmake b/cmake_modules/SetupCxxFlags.cmake
index 4df2f5601..855d8e1d3 100644
--- a/cmake_modules/SetupCxxFlags.cmake
+++ b/cmake_modules/SetupCxxFlags.cmake
@@ -22,6 +22,28 @@ include(CheckCXXCompilerFlag)
 
 message(STATUS "System processor: ${CMAKE_SYSTEM_PROCESSOR}")
 
+include("${CMAKE_CURRENT_LIST_DIR}/TargetArchitecture.cmake")
+message(STATUS "Target processor: ${PAIMON_TARGET_PROCESSOR} "
+               "(CPU family: ${PAIMON_TARGET_CPU_FAMILY})")
+
+# Probe the architecture specific tuning flags. x86 deliberately gets none of its
+# own, so there is nothing to probe there: the baseline stays whatever the
+# toolchain defaults to, and no persisted value depends on it; see
+# src/paimon/common/utils/crc32c.h.
+if(PAIMON_TARGET_CPU_FAMILY STREQUAL "aarch64" AND NOT "${PAIMON_AARCH64_MARCH}" STREQUAL
+                                                   "")
+    set(PAIMON_AARCH64_MARCH_FLAG "-march=${PAIMON_AARCH64_MARCH}")
+    # check_cxx_compiler_flag() caches under a fixed name, so a reconfigure in the
+    # same build directory would otherwise reuse the old PAIMON_AARCH64_MARCH verdict.
+    if(NOT "${PAIMON_AARCH64_MARCH_FLAG}" STREQUAL "${PAIMON_PROBED_AARCH64_MARCH_FLAG}")
+        unset(CXX_SUPPORTS_AARCH64_MARCH CACHE)
+        set(PAIMON_PROBED_AARCH64_MARCH_FLAG
+            "${PAIMON_AARCH64_MARCH_FLAG}"
+            CACHE INTERNAL "Arm64 arch flag that CXX_SUPPORTS_AARCH64_MARCH refers to")
+    endif()
+    check_cxx_compiler_flag("${PAIMON_AARCH64_MARCH_FLAG}" CXX_SUPPORTS_AARCH64_MARCH)
+endif()
+
 # Support C11
 if(NOT DEFINED CMAKE_C_STANDARD)
     set(CMAKE_C_STANDARD 11)
@@ -225,58 +247,22 @@ if(BUILD_WARNING_FLAGS)
 endif(BUILD_WARNING_FLAGS)
 
 # Only enable additional instruction sets if they are supported
-if(PAIMON_CPU_FLAG STREQUAL "x86")
-    if(PAIMON_SIMD_LEVEL STREQUAL "AVX512")
-        if(NOT CXX_SUPPORTS_AVX512)
-            message(FATAL_ERROR "AVX512 required but compiler doesn't support it.")
-        endif()
-        set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} ${PAIMON_AVX512_FLAG}")
-        add_definitions(-DPAIMON_HAVE_AVX512 -DPAIMON_HAVE_AVX2 -DPAIMON_HAVE_BMI2
-                        -DPAIMON_HAVE_SSE4_2)
-    elseif(PAIMON_SIMD_LEVEL STREQUAL "AVX2")
-        if(NOT CXX_SUPPORTS_AVX2)
-            message(FATAL_ERROR "AVX2 required but compiler doesn't support it.")
-        endif()
-        set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} ${PAIMON_AVX2_FLAG}")
-        add_definitions(-DPAIMON_HAVE_AVX2 -DPAIMON_HAVE_BMI2 -DPAIMON_HAVE_SSE4_2)
-    elseif(PAIMON_SIMD_LEVEL STREQUAL "SSE4_2")
-        if(NOT CXX_SUPPORTS_SSE4_2)
-            message(FATAL_ERROR "SSE4.2 required but compiler doesn't support it.")
-        endif()
-        set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} ${PAIMON_SSE4_2_FLAG}")
-        add_definitions(-DPAIMON_HAVE_SSE4_2)
-    endif()
-endif()
-
-if(PAIMON_CPU_FLAG STREQUAL "ppc")
-    if(CXX_SUPPORTS_ALTIVEC AND PAIMON_ALTIVEC)
-        set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} ${PAIMON_ALTIVEC_FLAG}")
-    endif()
-endif()
-
-if(PAIMON_CPU_FLAG STREQUAL "armv8")
-    if(NOT CXX_SUPPORTS_ARMV8_ARCH)
-        message(FATAL_ERROR "Unsupported arch flag: ${PAIMON_ARMV8_ARCH_FLAG}.")
-    endif()
-    if(PAIMON_ARMV8_ARCH_FLAG MATCHES "native")
-        message(FATAL_ERROR "native arch not allowed, please specify arch explicitly.")
-    endif()
-    set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} ${PAIMON_ARMV8_ARCH_FLAG}")
-
-    add_definitions(-DPAIMON_HAVE_NEON)
-
-    if(CMAKE_CXX_COMPILER_ID STREQUAL "GNU" AND CMAKE_CXX_COMPILER_VERSION VERSION_LESS
-                                                "5.4")
-        message(WARNING "Disable Armv8 CRC and Crypto as compiler doesn't support them well."
-        )
+if(PAIMON_TARGET_CPU_FAMILY STREQUAL "aarch64")
+    if("${PAIMON_AARCH64_MARCH}" STREQUAL "")
+        # An explicitly empty PAIMON_AARCH64_MARCH opts out of -march entirely, for
+        # toolchains that reject the flag, such as AppleClang targeting arm64.
+        message(STATUS "PAIMON_AARCH64_MARCH is empty: passing no -march flag")
     else()
-        if(PAIMON_ARMV8_ARCH_FLAG MATCHES "\\+crypto")
-            add_definitions(-DPAIMON_HAVE_ARMV8_CRYPTO)
+        if(NOT CXX_SUPPORTS_AARCH64_MARCH)
+            message(FATAL_ERROR "The compiler does not accept ${PAIMON_AARCH64_MARCH_FLAG}. "
+                                "Set -DPAIMON_AARCH64_MARCH to a value it accepts, or to an "
+                                "empty string to pass no -march flag.")
         endif()
-        # armv8.1+ implies crc support
-        if(PAIMON_ARMV8_ARCH_FLAG MATCHES "armv8\\.[1-9]|\\+crc")
-            add_definitions(-DPAIMON_HAVE_ARMV8_CRC)
+        if(PAIMON_AARCH64_MARCH_FLAG MATCHES "native")
+            message(FATAL_ERROR "native arch not allowed, please specify arch explicitly."
+            )
         endif()
+        set(CXX_COMMON_FLAGS "${CXX_COMMON_FLAGS} ${PAIMON_AARCH64_MARCH_FLAG}")
     endif()
 endif()
 
diff --git a/cmake_modules/TargetArchitecture.cmake b/cmake_modules/TargetArchitecture.cmake
new file mode 100644
index 000000000..8c5aaf315
--- /dev/null
+++ b/cmake_modules/TargetArchitecture.cmake
@@ -0,0 +1,50 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Resolves the target processor and the CPU family that architecture specific
+# build logic keys on. Deliberately free of compiler probes and of any state
+# beyond its inputs, so that cmake_modules/tests/target_architecture_test.cmake
+# can exercise it with `cmake -P`, without a toolchain.
+#
+# Inputs:  APPLE, CMAKE_SYSTEM_PROCESSOR, CMAKE_OSX_ARCHITECTURES
+# Outputs: PAIMON_TARGET_PROCESSOR, PAIMON_TARGET_CPU_FAMILY
+
+# Resolve the CMake-declared target processor. Only on Apple platforms does
+# CMAKE_OSX_ARCHITECTURES describe the target -- elsewhere a stale or inherited
+# value must not be read. A universal binary names several architectures at once,
+# so it gets no architecture specific tuning.
+if(APPLE)
+    list(LENGTH CMAKE_OSX_ARCHITECTURES _paimon_osx_arch_count)
+else()
+    set(_paimon_osx_arch_count 0)
+endif()
+if(_paimon_osx_arch_count EQUAL 1)
+    set(PAIMON_TARGET_PROCESSOR "${CMAKE_OSX_ARCHITECTURES}")
+elseif(_paimon_osx_arch_count EQUAL 0)
+    set(PAIMON_TARGET_PROCESSOR "${CMAKE_SYSTEM_PROCESSOR}")
+else()
+    set(PAIMON_TARGET_PROCESSOR "")
+endif()
+
+# Unrecognized processors are not an error; they just get no tuning flags.
+if(PAIMON_TARGET_PROCESSOR MATCHES "^(x86_64|amd64|AMD64)$")
+    set(PAIMON_TARGET_CPU_FAMILY "x86")
+elseif(PAIMON_TARGET_PROCESSOR MATCHES "^(aarch64|arm64|ARM64)$")
+    set(PAIMON_TARGET_CPU_FAMILY "aarch64")
+else()
+    set(PAIMON_TARGET_CPU_FAMILY "unknown")
+endif()
diff --git a/cmake_modules/ThirdpartyToolchain.cmake b/cmake_modules/ThirdpartyToolchain.cmake
index ce7978327..bd3ed2a05 100644
--- a/cmake_modules/ThirdpartyToolchain.cmake
+++ b/cmake_modules/ThirdpartyToolchain.cmake
@@ -1261,7 +1261,27 @@ macro(build_jindosdk_c)
 endmacro()
 
 macro(build_lumina)
-    message(STATUS "Installing Lumina from precompiled package")
+    # The release tarball ships one prebuilt artifact tree per platform; keep this
+    # list in sync with the platforms the pinned release actually contains.
+    set(LUMINA_SUPPORTED_PLATFORMS "linux-x86_64")
+    if(CMAKE_SYSTEM_NAME STREQUAL "Linux" AND PAIMON_TARGET_CPU_FAMILY STREQUAL "x86")
+        set(LUMINA_ARTIFACT_PLATFORM "linux-x86_64")
+    elseif(CMAKE_SYSTEM_NAME STREQUAL "Linux" AND PAIMON_TARGET_CPU_FAMILY STREQUAL
+                                                  "aarch64")
+        set(LUMINA_ARTIFACT_PLATFORM "linux-aarch64")
+    else()
+        set(LUMINA_ARTIFACT_PLATFORM "${CMAKE_SYSTEM_NAME}-${PAIMON_TARGET_PROCESSOR}")
+    endif()
+    if(NOT "${LUMINA_ARTIFACT_PLATFORM}" IN_LIST LUMINA_SUPPORTED_PLATFORMS)
+        message(FATAL_ERROR "Lumina ${PAIMON_LUMINA_BUILD_VERSION} has no prebuilt artifacts for "
+                            "${LUMINA_ARTIFACT_PLATFORM} (available: ${LUMINA_SUPPORTED_PLATFORMS}). "
+                            "Reconfigure with -DPAIMON_ENABLE_LUMINA=OFF.")
+    endif()
+    set(LUMINA_ARTIFACT_DIR
+        "artifacts/cpp/${LUMINA_ARTIFACT_PLATFORM}/install-root/usr/local")
+
+    message(STATUS "Installing Lumina for ${LUMINA_ARTIFACT_PLATFORM} from precompiled package"
+    )
 
     set(LUMINA_PREFIX "${CMAKE_CURRENT_BINARY_DIR}/lumina_ep-install")
     set(LUMINA_INCLUDE_DIR "${LUMINA_PREFIX}/include")
@@ -1275,10 +1295,10 @@ macro(build_lumina)
                         CONFIGURE_COMMAND ""
                         BUILD_COMMAND ""
                         INSTALL_COMMAND ${CMAKE_COMMAND} -E copy_directory
-                                        <SOURCE_DIR>/artifacts/cpp/linux-x86_64/install-root/usr/local/include
+                                        <SOURCE_DIR>/${LUMINA_ARTIFACT_DIR}/include
                                         ${LUMINA_INCLUDE_DIR}
                         COMMAND ${CMAKE_COMMAND} -E copy_if_different
-                                <SOURCE_DIR>/artifacts/cpp/linux-x86_64/install-root/usr/local/lib/liblumina.so
+                                <SOURCE_DIR>/${LUMINA_ARTIFACT_DIR}/lib/liblumina.so
                                 ${LUMINA_DYNAMIC_LIB}
                         BUILD_BYPRODUCTS "${LUMINA_DYNAMIC_LIB}")
 
diff --git a/cmake_modules/san-config.cmake b/cmake_modules/san-config.cmake
index a2ab16aed..6392a7bc6 100644
--- a/cmake_modules/san-config.cmake
+++ b/cmake_modules/san-config.cmake
@@ -51,6 +51,28 @@ if(PAIMON_USE_UBSAN)
                                INTERFACE -fsanitize=undefined -fno-sanitize=vptr
                                          -fno-omit-frame-pointer)
         target_link_options(paimon_sanitizer_flags INTERFACE -fsanitize=undefined)
+        # The signed-integer-overflow check on a 128-bit multiplication becomes a call to
+        # __muloti4, which Clang takes from compiler-rt. libgcc does not provide it, and on
+        # aarch64 Clang does not inline the check the way it can on x86-64, so the link needs
+        # compiler-rt's builtins. Only that archive is added, rather than switching the whole
+        # runtime library with --rtlib=compiler-rt. GCC lowers the check through libgcc and
+        # never emits __muloti4, and it rejects the --rtlib= driver flag, so the probe is
+        # Clang only.
+        if(CMAKE_CXX_COMPILER_ID MATCHES "Clang")
+            execute_process(COMMAND ${CMAKE_CXX_COMPILER} --rtlib=compiler-rt
+                                    --print-libgcc-file-name
+                            OUTPUT_VARIABLE PAIMON_COMPILER_RT_BUILTINS
+                            OUTPUT_STRIP_TRAILING_WHITESPACE ERROR_QUIET)
+            if(EXISTS "${PAIMON_COMPILER_RT_BUILTINS}")
+                target_link_libraries(paimon_sanitizer_flags
+                                      INTERFACE "${PAIMON_COMPILER_RT_BUILTINS}")
+                message(STATUS "Undefined Behavior Sanitizer builtins: ${PAIMON_COMPILER_RT_BUILTINS}"
+                )
+            else()
+                message(WARNING "compiler-rt builtins not found; a 128-bit multiplication under "
+                                "-fsanitize=undefined may fail to link")
+            endif()
+        endif()
         message(STATUS "Undefined Behavior Sanitizer enabled")
     else()
         message(WARNING "Undefined Behavior Sanitizer is only supported for GCC and Clang compilers"
diff --git a/cmake_modules/tests/aarch64_march_option_test.cmake b/cmake_modules/tests/aarch64_march_option_test.cmake
new file mode 100644
index 000000000..5bbda9054
--- /dev/null
+++ b/cmake_modules/tests/aarch64_march_option_test.cmake
@@ -0,0 +1,115 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Regression tests for how cmake_modules/DefineOptions.cmake resolves
+# PAIMON_AARCH64_MARCH. On AArch64 targets SetupCxxFlags.cmake turns it into
+# "-march=${PAIMON_AARCH64_MARCH}", except for an explicitly empty value, the
+# opt-out it answers with no -march flag at all. Only an undefined value may fall
+# back to the default: clobbering the empty opt-out would put the flag back. The
+# options themselves are only defined for a top-level build, which makes the
+# add_subdirectory() path the interesting one.
+#
+#   cmake -P cmake_modules/tests/aarch64_march_option_test.cmake
+#
+# Script mode does have a cache: `cmake -D... -P` populates one, and
+# set(... CACHE ...) defers to it exactly as a real configure does. That makes the
+# -D override testable, and it also means each case has to clear the cache entry as
+# well as the normal variable, so an ambient -DPAIMON_AARCH64_MARCH cannot change
+# the outcome.
+
+# Pin the policies the real build runs under: the top-level CMakeLists.txt
+# requires 3.22, so CMP0126 is NEW there and set(CACHE) leaves a normal variable
+# alone. Script mode would otherwise apply the OLD behavior and erase the
+# explicitly empty opt-out value.
+cmake_minimum_required(VERSION 3.22)
+
+set(DEFINE_OPTIONS_MODULE "${CMAKE_CURRENT_LIST_DIR}/../DefineOptions.cmake")
+set(FAILURES 0)
+set(CHECKS 0)
+
+# A macro, not a function, so FAILURES accumulates in this scope. Pass <unset> as
+# the preset to leave PAIMON_AARCH64_MARCH undefined going in.
+macro(expect_march
+      description
+      top_level
+      preset
+      expected)
+    if(${top_level})
+        set(CMAKE_SOURCE_DIR "/paimon")
+        set(CMAKE_CURRENT_SOURCE_DIR "/paimon")
+    else()
+        set(CMAKE_SOURCE_DIR "/superproject")
+        set(CMAKE_CURRENT_SOURCE_DIR "/superproject/paimon")
+    endif()
+
+    unset(PAIMON_AARCH64_MARCH)
+    unset(PAIMON_AARCH64_MARCH CACHE)
+    if(NOT "${preset}" STREQUAL "<unset>")
+        set(PAIMON_AARCH64_MARCH "${preset}")
+    endif()
+
+    include("${DEFINE_OPTIONS_MODULE}")
+
+    math(EXPR CHECKS "${CHECKS} + 1")
+    if(NOT PAIMON_AARCH64_MARCH STREQUAL "${expected}")
+        math(EXPR FAILURES "${FAILURES} + 1")
+        message(SEND_ERROR "${description}: expected '${expected}', got "
+                           "'${PAIMON_AARCH64_MARCH}'")
+    endif()
+endmacro()
+
+# What -DPAIMON_AARCH64_MARCH=... produces on a top-level build: the cache entry
+# already exists, so define_option_string must not overwrite it and the fallback
+# must not clobber it either.
+macro(expect_march_cached description value expected)
+    set(CMAKE_SOURCE_DIR "/paimon")
+    set(CMAKE_CURRENT_SOURCE_DIR "/paimon")
+    unset(PAIMON_AARCH64_MARCH)
+    unset(PAIMON_AARCH64_MARCH CACHE)
+    set(PAIMON_AARCH64_MARCH
+        "${value}"
+        CACHE STRING "" FORCE)
+
+    include("${DEFINE_OPTIONS_MODULE}")
+
+    math(EXPR CHECKS "${CHECKS} + 1")
+    if(NOT PAIMON_AARCH64_MARCH STREQUAL "${expected}")
+        math(EXPR FAILURES "${FAILURES} + 1")
+        message(SEND_ERROR "${description}: expected '${expected}', got "
+                           "'${PAIMON_AARCH64_MARCH}'")
+    endif()
+endmacro()
+
+# The option is never defined on this path, so only the fallback can supply a value.
+expect_march("add_subdirectory consumer gets the default" FALSE "<unset>" "armv8-a")
+expect_march("add_subdirectory keeps an explicit value" FALSE "armv8.2-a+crc"
+             "armv8.2-a+crc")
+
+# An explicitly empty value is the opt-out and must survive the fallback.
+expect_march("an empty value survives as the opt-out, off the top level" FALSE "" "")
+expect_march("an empty value survives as the opt-out, on the top level" TRUE "" "")
+
+expect_march("top-level build gets the default" TRUE "<unset>" "armv8-a")
+
+expect_march_cached("a -D override survives" "armv9-a" "armv9-a")
+expect_march_cached("a -D override may add extensions" "armv8.2-a+crc" "armv8.2-a+crc")
+expect_march_cached("a -D empty override survives as the opt-out" "" "")
+
+if(FAILURES GREATER 0)
+    message(FATAL_ERROR "${FAILURES} of ${CHECKS} AArch64 -march checks failed")
+endif()
+message(STATUS "All ${CHECKS} AArch64 -march checks passed")
diff --git a/cmake_modules/tests/target_architecture_test.cmake b/cmake_modules/tests/target_architecture_test.cmake
new file mode 100644
index 000000000..8a0c69103
--- /dev/null
+++ b/cmake_modules/tests/target_architecture_test.cmake
@@ -0,0 +1,114 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+# Regression tests for cmake_modules/TargetArchitecture.cmake. No toolchain is
+# needed, so this runs anywhere CMake does:
+#
+#   cmake -P cmake_modules/tests/target_architecture_test.cmake
+
+set(TARGET_ARCHITECTURE_MODULE "${CMAKE_CURRENT_LIST_DIR}/../TargetArchitecture.cmake")
+set(FAILURES 0)
+set(CHECKS 0)
+
+# Macros, not functions, so FAILURES accumulates in this scope. Every branch of the
+# module assigns both outputs, so cases cannot leak into one another. The case
+# helpers take at most four positional arguments to stay within the repository
+# cmake-format profile, which would otherwise wrap every call onto six lines.
+macro(run_case)
+    set(PAIMON_TARGET_PROCESSOR "<unset>")
+    set(PAIMON_TARGET_CPU_FAMILY "<unset>")
+
+    include("${TARGET_ARCHITECTURE_MODULE}")
+
+    math(EXPR CHECKS "${CHECKS} + 1")
+    if(NOT PAIMON_TARGET_PROCESSOR STREQUAL "${EXPECTED_PROCESSOR}"
+       OR NOT PAIMON_TARGET_CPU_FAMILY STREQUAL "${EXPECTED_FAMILY}")
+        math(EXPR FAILURES "${FAILURES} + 1")
+        message(SEND_ERROR "${CASE}: expected processor '${EXPECTED_PROCESSOR}' family "
+                           "'${EXPECTED_FAMILY}', got processor '${PAIMON_TARGET_PROCESSOR}' "
+                           "family '${PAIMON_TARGET_CPU_FAMILY}'")
+    endif()
+endmacro()
+
+# A non-Apple target. CMAKE_OSX_ARCHITECTURES must be ignored there, so the
+# resolved processor is always CMAKE_SYSTEM_PROCESSOR.
+macro(expect_non_apple
+      description
+      processor
+      osx_archs
+      family)
+    set(CASE "${description}")
+    set(APPLE FALSE)
+    set(CMAKE_SYSTEM_PROCESSOR "${processor}")
+    set(CMAKE_OSX_ARCHITECTURES "${osx_archs}")
+    set(EXPECTED_PROCESSOR "${processor}")
+    set(EXPECTED_FAMILY "${family}")
+    run_case()
+endmacro()
+
+# An Apple target, where CMAKE_OSX_ARCHITECTURES decides and may disagree with the
+# host processor.
+macro(expect_apple
+      host_processor
+      osx_archs
+      expected_processor
+      family)
+    set(CASE "apple host='${host_processor}' osx='${osx_archs}'")
+    set(APPLE TRUE)
+    set(CMAKE_SYSTEM_PROCESSOR "${host_processor}")
+    set(CMAKE_OSX_ARCHITECTURES "${osx_archs}")
+    set(EXPECTED_PROCESSOR "${expected_processor}")
+    set(EXPECTED_FAMILY "${family}")
+    run_case()
+endmacro()
+
+# x86_64 spellings across Linux, the BSDs and Windows.
+expect_non_apple("linux x86_64" "x86_64" "" "x86")
+expect_non_apple("lowercase amd64" "amd64" "" "x86")
+expect_non_apple("windows AMD64" "AMD64" "" "x86")
+
+# Arm64 spellings: Linux reports aarch64, Apple and Windows report arm64/ARM64.
+expect_non_apple("linux aarch64" "aarch64" "" "aarch64")
+expect_non_apple("lowercase arm64" "arm64" "" "aarch64")
+expect_non_apple("windows ARM64" "ARM64" "" "aarch64")
+
+# Architectures without Paimon tuning must configure, not fail.
+expect_non_apple("ppc64le is not an error" "ppc64le" "" "unknown")
+expect_non_apple("s390x is not an error" "s390x" "" "unknown")
+expect_non_apple("riscv64 is not an error" "riscv64" "" "unknown")
+expect_non_apple("32-bit arm is not aarch64" "armv7l" "" "unknown")
+expect_non_apple("empty processor" "" "" "unknown")
+
+# CMAKE_OSX_ARCHITECTURES describes the target only on Apple platforms. A value
+# inherited from a superproject or left in the cache must not retarget a Linux
+# build, which would otherwise add an Arm -march= flag to an x86_64 compiler.
+expect_non_apple("linux ignores stale osx arm64" "x86_64" "arm64" "x86")
+expect_non_apple("linux ignores stale osx universal" "x86_64" "arm64;x86_64" "x86")
+
+# On Apple the compiler targets CMAKE_OSX_ARCHITECTURES, which may disagree with
+# the host CMAKE_SYSTEM_PROCESSOR.
+expect_apple("arm64" "" "arm64" "aarch64")
+expect_apple("x86_64" "arm64" "arm64" "aarch64")
+expect_apple("arm64" "x86_64" "x86_64" "x86")
+
+# A universal binary targets several architectures, so none of them may be tuned for.
+expect_apple("arm64" "arm64;x86_64" "" "unknown")
+
+if(FAILURES GREATER 0)
+    message(FATAL_ERROR "${FAILURES} of ${CHECKS} target architecture checks failed")
+endif()
+message(STATUS "All ${CHECKS} target architecture checks passed")
diff --git a/docs/code-style.md b/docs/code-style.md
index 1806dc939..672cda518 100644
--- a/docs/code-style.md
+++ b/docs/code-style.md
@@ -27,7 +27,25 @@ This document defines the coding conventions for the Paimon C++ project. All pul
 
 - **C++17** is the target standard.
 - Do **not** use C++20 or later features.
-- Only the **x86_64** architecture is currently supported.
+- The **x86_64** and **aarch64** architectures are supported. Do not assume a specific
+  architecture.
+- Never rely on the signedness of plain `char`. It is
+  [implementation defined](https://gcc.gnu.org/onlinedocs/gcc/Characters-implementation.html),
+  is fixed by the target ABI rather than by the architecture as such (the AArch64 and
+  x86-64 psABIs disagree), and can be flipped with
+  `-fsigned-char` / `-funsigned-char`. Use `int8_t` or `uint8_t` when the signedness of a
+  byte matters, and keep `char` for text.
+- Converting a floating point value to an integer type is
+  [undefined behavior](https://en.cppreference.com/w/cpp/language/implicit_conversion) unless the
+  truncated value is representable in the target type; `NaN` and the infinities never are. Check
+  the value for finiteness and range first, then apply an explicit policy — reject it or saturate
+  it. Do not rely on what a given architecture happens to produce, and do not let a test reach the
+  conversion with an unrepresentable value: Clang's `-fsanitize=float-cast-overflow` reports it.
+  `NumericPrimitiveCastExecutor::JavaFloatingToIntegerCast` implements the policy Paimon uses,
+  which is the one Java applies: `NaN` becomes 0, and an out of range value saturates at the int32
+  bounds, or the int64 bounds for a BIGINT destination, and is then narrowed to the target width by
+  keeping the low bits. The narrowing is why `300.9` converts to 44 as a TINYINT — the low byte of
+  300 — and saturating first is why `MAX_FLOAT` converts to -1 rather than to 127.
 
 ---
 
diff --git a/docs/source/building.rst b/docs/source/building.rst
index 0c5ece36b..466d461b4 100644
--- a/docs/source/building.rst
+++ b/docs/source/building.rst
@@ -33,7 +33,7 @@ out-of-source. For example, you could create ``paimon-cpp/build`` and invoke
 Building requires:
 
 * A C++17-enabled compiler. On Linux, gcc 8 and higher should be sufficient.
-  macOS and Windows are not supported for now.
+  Windows is not supported for now.
 * At least 2GB of RAM for a minimal build, 8GB for a minimal
   debug build with tests and 16GB for a full build.
 
@@ -48,6 +48,61 @@ On Ubuntu/Debian you can install the requirements with:
 We also provide a docker template to help you get started quickly. See in
 ``.devcontainer`` folder for more details.
 
+.. _cpp-building-platforms:
+
+Supported platforms
+-------------------
+
+.. list-table::
+   :header-rows: 1
+   :widths: 20 25 55
+
+   * - Operating system
+     - Architecture
+     - Support level
+   * - Linux
+     - ``x86_64``
+     - Built and tested in CI.
+   * - Linux
+     - ``aarch64`` / ``arm64``
+     - Built and tested in CI.
+   * - macOS
+     - ``x86_64``, ``arm64``
+     - Architecture detection is wired up, but macOS as a whole is not an
+       officially supported platform and is not covered by CI.
+   * - Windows
+     - any
+     - Not supported.
+
+The build system derives the target CPU family from
+``CMAKE_SYSTEM_PROCESSOR`` (or from ``CMAKE_OSX_ARCHITECTURES`` on macOS) and
+only applies tuning flags belonging to that family. On an architecture it does
+not recognise it simply adds no tuning flags of its own, which is not by itself
+a statement that the dependencies or the runtime work there. On aarch64 the
+``-march=`` value defaults to ``armv8-a`` and can be overridden with
+``-DPAIMON_AARCH64_MARCH=armv8.2-a+crc`` or similar; an explicitly empty value
+(``-DPAIMON_AARCH64_MARCH=""``) passes no ``-march`` flag at all, for toolchains
+that reject the flag, such as AppleClang targeting arm64.
+
+There is no x86 SIMD level to select, and no instruction-set flag is added on
+x86: the baseline stays whatever the toolchain defaults to. No persisted value
+depends on it, in particular the checksum written to SST files and to the B-tree
+global index is the same however Paimon was compiled.
+
+One optional component is architecture-restricted:
+``-DPAIMON_ENABLE_LUMINA=ON`` consumes a prebuilt binary that is only
+published for Linux ``x86_64``, so configuring it on any other platform fails
+with an explicit error.
+
+Cross compilation is not wired up end to end: the bundled third-party builds are
+handed only the C and C++ compilers, without a toolchain file, sysroot or find
+root, so they can still pick up host libraries. Build natively on each target.
+
+``build_and_package.sh`` therefore names its output after the host platform, for
+example ``output/paimon-cpp-linux-aarch64.tar.gz``, so artifacts built on several
+machines can be collected side by side. ``--platform <label>`` overrides that
+label alone; it does not configure a cross build.
+
 .. _cpp-building-building:
 
 Building
@@ -123,7 +178,8 @@ boolean flags to ``cmake``.
 * ``-DPAIMON_ENABLE_ORC=ON``: Paimon integration with Apache ORC
 * ``-DPAIMON_ENABLE_AVRO=ON``: Apache Avro libraries and Paimon integration
 * ``-DPAIMON_ENABLE_JINDO=ON``: Support for Alibaba Jindo filesystems
-* ``-DPAIMON_ENABLE_LUMINA=ON``: Support for the Lumina vector index.
+* ``-DPAIMON_ENABLE_LUMINA=ON``: Support for the Lumina vector index. Requires
+  Linux ``x86_64``; see :ref:`cpp-building-platforms`.
 * ``-DPAIMON_ENABLE_LUCENE=ON``: Support for Lucene full-text search indexes
 * ``-DPAIMON_ENABLE_TANTIVY=ON``: Enable the experimental Tantivy full-text index Rust FFI.
 * ``-DPAIMON_ENABLE_REST=ON``: Support for the REST catalog (``metastore=rest``), requires the libcurl development package.
diff --git a/docs/source/user_guide/read.rst b/docs/source/user_guide/read.rst
index 8d852d920..d628fc1ec 100644
--- a/docs/source/user_guide/read.rst
+++ b/docs/source/user_guide/read.rst
@@ -80,6 +80,9 @@ Overflow behavior is undefined for C++ and Java Paimon. Results in overflow scen
 Paimon C++ does not guarantee identical results to Java Paimon in overflow scenarios. Users should not rely on identical
 return values between implementations.
 
+One exception: casting ``float`` or ``double`` to an integer type is well defined and matches Java on every supported
+architecture; see note 2️⃣ below the matrix.
+
 Type Change Support Matrix
 ~~~~~~~~~~~~~~~~~~~~~~~~~~
 The table below indicates support for changing a column type from ``source`` to ``target``. Refer to the numbered notes below the table
@@ -265,10 +268,15 @@ for caveats.
   1️⃣ Integer downcast overflow behavior matches Java in specific cases.
     Example: smallint -> tinyint, 32767 becomes -1; int -> smallint, -2147483648 becomes 0.
 
-  2️⃣ Floating-point overflow behavior is partially consistent with Java and partially different.
-    Example: float -> tinyint
-      - Java: MAX_FLOAT -> -1, INFINITY -> -1
-      - C++:  MAX_FLOAT -> 0, INFINITY -> 0
+  2️⃣ Casting float/double to an integer type follows Java semantics on every supported architecture:
+    ``NaN`` becomes 0, an out-of-range value saturates at the int32 bounds (int64 for ``bigint``),
+    and the result is narrowed to the target width by keeping the low bits, the way Java
+    narrowing does.
+
+    Example: float -> tinyint, C++ and Java both: ``MAX_FLOAT -> -1``, ``INFINITY -> -1``,
+    ``NaN -> 0``, ``300.9 -> 44``.
+
+    double -> float overflow produces ``Infinity`` / ``-Infinity``, consistent with Java.
 
   3️⃣ Keyword differences for special float/double values:
     - Java: Infinity, -Infinity, NaN
diff --git a/src/paimon/common/data/binary_string.cpp b/src/paimon/common/data/binary_string.cpp
index 5dc56b3e9..b8880456a 100644
--- a/src/paimon/common/data/binary_string.cpp
+++ b/src/paimon/common/data/binary_string.cpp
@@ -21,6 +21,7 @@
 
 #include <algorithm>
 #include <cctype>
+#include <cstdint>
 
 #include "paimon/common/memory/memory_segment_utils.h"
 #include "paimon/memory/bytes.h"
@@ -75,16 +76,19 @@ std::string BinaryString::ToString() const {
 }
 
 int32_t BinaryString::NumBytesForFirstByte(char b) {
-    if (b >= 0) {
+    // Classify the leading byte as unsigned: where the ABI makes plain char unsigned, as the
+    // AArch64 Linux ABI does, a signed test would classify every byte as single byte ASCII.
+    const auto first_byte = static_cast<uint8_t>(b);
+    if (first_byte < 0x80) {
         // 1 byte, 7 bits: 0xxxxxxx
         return 1;
-    } else if ((b >> 5) == -2 && (b & 0x1e) != 0) {
+    } else if ((first_byte & 0xe0) == 0xc0 && (first_byte & 0x1e) != 0) {
         // 2 bytes, 11 bits: 110xxxxx 10xxxxxx
         return 2;
-    } else if ((b >> 4) == -2) {
+    } else if ((first_byte & 0xf0) == 0xe0) {
         // 3 bytes, 16 bits: 1110xxxx 10xxxxxx 10xxxxxx
         return 3;
-    } else if ((b >> 3) == -2) {
+    } else if ((first_byte & 0xf8) == 0xf0) {
         // 4 bytes, 21 bits: 11110xxx 10xxxxxx 10xxxxxx 10xxxxxx
         return 4;
     } else {
@@ -202,14 +206,15 @@ BinaryString BinaryString::ToUpperCase(MemoryPool* pool) const {
     int32_t size = segment_.Size();
     SegmentAndOffset segment_and_offset = StartSegmentAndOffset(size);
     std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(size_in_bytes_, pool);
-    (*bytes)[0] = tolower(segment_and_offset.Value());
     for (int32_t i = 0; i < size_in_bytes_; i++) {
         char b = segment_and_offset.Value();
         if (NumBytesForFirstByte(b) != 1) {
             // fallback
             return CppToUpperCase(pool);
         }
-        int32_t upper = toupper(static_cast<int32_t>(b));
+        // A byte >= 0x80 still reaches here via the invalid-first-byte path, and toupper()
+        // requires a non-negative argument, which plain char does not guarantee.
+        int32_t upper = toupper(static_cast<unsigned char>(b));
         if (upper > 127) {
             // fallback
             return CppToUpperCase(pool);
@@ -222,7 +227,8 @@ BinaryString BinaryString::ToUpperCase(MemoryPool* pool) const {
 
 BinaryString BinaryString::CppToUpperCase(MemoryPool* pool) const {
     std::string str = ToString();
-    std::transform(str.begin(), str.end(), str.begin(), ::toupper);
+    std::transform(str.begin(), str.end(), str.begin(),
+                   [](unsigned char c) { return static_cast<char>(::toupper(c)); });
     return FromString(str, pool);
 }
 
@@ -233,14 +239,14 @@ BinaryString BinaryString::ToLowerCase(MemoryPool* pool) const {
     int32_t size = segment_.Size();
     SegmentAndOffset segment_and_offset = StartSegmentAndOffset(size);
     std::shared_ptr<Bytes> bytes = Bytes::AllocateBytes(size_in_bytes_, pool);
-    (*bytes)[0] = tolower(segment_and_offset.Value());
     for (int32_t i = 0; i < size_in_bytes_; i++) {
         char b = segment_and_offset.Value();
         if (NumBytesForFirstByte(b) != 1) {
             // fallback
             return CppToLowerCase(pool);
         }
-        int32_t lower = tolower(static_cast<int32_t>(b));
+        // As in ToUpperCase: a byte >= 0x80 must stay non-negative for tolower().
+        int32_t lower = tolower(static_cast<unsigned char>(b));
         if (lower > 127) {
             // fallback
             return CppToLowerCase(pool);
@@ -253,7 +259,8 @@ BinaryString BinaryString::ToLowerCase(MemoryPool* pool) const {
 
 BinaryString BinaryString::CppToLowerCase(MemoryPool* pool) const {
     std::string str = ToString();
-    std::transform(str.begin(), str.end(), str.begin(), ::tolower);
+    std::transform(str.begin(), str.end(), str.begin(),
+                   [](unsigned char c) { return static_cast<char>(::tolower(c)); });
     return FromString(str, pool);
 }
 
diff --git a/src/paimon/common/data/binary_string_test.cpp b/src/paimon/common/data/binary_string_test.cpp
index 797895bcc..5c87ca7b2 100644
--- a/src/paimon/common/data/binary_string_test.cpp
+++ b/src/paimon/common/data/binary_string_test.cpp
@@ -83,10 +83,11 @@ TEST_F(BinaryStringTest, TestBasic) {
     CheckBasic("Paimon中文社区", 10);
     CheckBasic("中 文 社 区", 7);
 
-    CheckBasic("¡", 1);       // 2 bytes char
-    CheckBasic("ку", 2);      // 2 * 2 bytes chars
-    CheckBasic("︽﹋％", 3);  // 3 * 3 bytes chars
-    // CheckBasic("\uD83E\uDD19", 1);  // 4 bytes char
+    CheckBasic("¡", 1);             // 2 bytes char
+    CheckBasic("ку", 2);            // 2 * 2 bytes chars
+    CheckBasic("︽﹋％", 3);        // 3 * 3 bytes chars
+    CheckBasic(u8"\U0001F919", 1);  // 4 bytes char, written as a code point rather than
+                                    // as a surrogate pair, which is ill-formed in C++
 }
 
 TEST_F(BinaryStringTest, EmptyStringTest) {
@@ -196,6 +197,9 @@ TEST_F(BinaryStringTest, TestSubstring) {
     InnerCheckEqual(FromString("数据砖头").Substring(1, 3, pool.get()), FromString("据砖"));
     InnerCheckEqual(FromString("数据砖头").Substring(3, 5, pool.get()), FromString("头"));
     InnerCheckEqual(FromString("ߵ梷").Substring(0, 2, pool.get()), FromString("ߵ梷"));
+    InnerCheckEqual(FromString(u8"\U0001F919ab").Substring(0, 1, pool.get()),
+                    FromString(u8"\U0001F919"));
+    InnerCheckEqual(FromString(u8"\U0001F919ab").Substring(1, 3, pool.get()), FromString("ab"));
 }
 
 TEST_F(BinaryStringTest, TestSubStringAndCopyBinaryString) {
@@ -229,6 +233,8 @@ TEST_F(BinaryStringTest, TestIndexOf) {
         InnerCheckEqual(FromString("数据砖头").IndexOf(FromString("数"), 3), -1);
         InnerCheckEqual(FromString("数据砖头").IndexOf(FromString("数"), 0), 0);
         InnerCheckEqual(FromString("数据砖头").IndexOf(FromString("头"), 0), 3);
+        InnerCheckEqual(FromString(u8"a\U0001F919b").IndexOf(FromString(u8"\U0001F919"), 0), 1);
+        InnerCheckEqual(FromString(u8"a\U0001F919b").IndexOf(FromString("b"), 0), 2);
     }
     {
         auto pool = GetDefaultPool();
@@ -257,6 +263,12 @@ TEST_F(BinaryStringTest, TestToUpperLowerCase) {
     InnerCheckEqual(FromString("!@#$%^*").ToLowerCase(pool.get()), FromString("!@#$%^*"));
     InnerCheckEqual(FromString("!@#$%^*").ToLowerCase(pool.get()), FromString("!@#$%^*"));
     InnerCheckEqual(BinaryString::EmptyUtf8().ToLowerCase(pool.get()), BinaryString::EmptyUtf8());
+
+    // Invalid UTF-8 bytes pass through case conversion unchanged on both char-signedness ABIs.
+    InnerCheckEqual(FromString("a\x80z").ToUpperCase(pool.get()), FromString("A\x80Z"));
+    InnerCheckEqual(FromString("A\x80Z").ToLowerCase(pool.get()), FromString("a\x80z"));
+    InnerCheckEqual(FromString("ab\xff").ToUpperCase(pool.get()), FromString("AB\xff"));
+    InnerCheckEqual(FromString("AB\xff").ToLowerCase(pool.get()), FromString("ab\xff"));
 }
 
 TEST_F(BinaryStringTest, TestEmptyString) {
diff --git a/src/paimon/common/data/data_define.h b/src/paimon/common/data/data_define.h
index 6fa3f45c7..376c0de5d 100644
--- a/src/paimon/common/data/data_define.h
+++ b/src/paimon/common/data/data_define.h
@@ -108,6 +108,10 @@ class DataDefine {
                     return "map";
                 } else if constexpr (std::is_same_v<T, std::string_view>) {
                     return std::string(arg);
+                } else if constexpr (std::is_same_v<T, char>) {
+                    // TINYINT is held as a plain char, whose signedness follows the ABI, and
+                    // std::to_string would print 236 rather than -20 where it is unsigned.
+                    return std::to_string(static_cast<int32_t>(static_cast<int8_t>(arg)));
                 } else {
                     return std::to_string(arg);
                 }
diff --git a/src/paimon/common/data/data_define_test.cpp b/src/paimon/common/data/data_define_test.cpp
index 2a7683796..04d1e6f92 100644
--- a/src/paimon/common/data/data_define_test.cpp
+++ b/src/paimon/common/data/data_define_test.cpp
@@ -97,6 +97,17 @@ TEST(DataDefineTest, VariantValueToStringReturnsStringForInt) {
     ASSERT_EQ(DataDefine::VariantValueToString(int_variant), "42");
 }
 
+// Test case: VariantValueToString should print a negative TINYINT as a negative number, which
+// it holds as a plain char whose signedness follows the ABI
+TEST(DataDefineTest, VariantValueToStringReturnsSignedStringForTinyInt) {
+    VariantType tiny_int_variant = static_cast<char>(-20);
+    ASSERT_EQ(DataDefine::VariantValueToString(tiny_int_variant), "-20");
+    tiny_int_variant = static_cast<char>(127);
+    ASSERT_EQ(DataDefine::VariantValueToString(tiny_int_variant), "127");
+    tiny_int_variant = static_cast<char>(-128);
+    ASSERT_EQ(DataDefine::VariantValueToString(tiny_int_variant), "-128");
+}
+
 // Test case: VariantValueToString should handle string data (BinaryString)
 TEST(DataDefineTest, VariantValueToStringReturnsStringForBinaryString) {
     auto pool = GetDefaultPool();
diff --git a/src/paimon/common/utils/binary_row_partition_computer_test.cpp b/src/paimon/common/utils/binary_row_partition_computer_test.cpp
index 0bbe5e272..974c4d7af 100644
--- a/src/paimon/common/utils/binary_row_partition_computer_test.cpp
+++ b/src/paimon/common/utils/binary_row_partition_computer_test.cpp
@@ -77,7 +77,9 @@ TEST(BinaryRowPartitionComputerTest, TestToAndFromBinaryRow) {
         ASSERT_OK_AND_ASSIGN(BinaryRow row, computer->ToBinaryRow(partition_map));
         ASSERT_EQ(12, row.GetFieldCount());
         ASSERT_EQ(true, row.GetBoolean(0));
-        ASSERT_EQ(-20, row.GetByte(1));
+        // GetByte() returns a plain char, whose signedness the target ABI decides and
+        // -fsigned-char / -funsigned-char can flip; the AArch64 Linux ABI makes it unsigned.
+        ASSERT_EQ(-20, static_cast<int8_t>(row.GetByte(1)));
         ASSERT_EQ(10, row.GetByte(2));
         ASSERT_EQ(1556, row.GetShort(3));
         ASSERT_EQ(-2556, row.GetShort(4));
@@ -121,7 +123,7 @@ TEST(BinaryRowPartitionComputerTest, TestToAndFromBinaryRow) {
         ASSERT_OK_AND_ASSIGN(BinaryRow row, computer->ToBinaryRow(partition_map));
         ASSERT_EQ(12, row.GetFieldCount());
         ASSERT_EQ(true, row.GetBoolean(0));
-        ASSERT_EQ(-20, row.GetByte(1));
+        ASSERT_EQ(-20, static_cast<int8_t>(row.GetByte(1)));
         ASSERT_EQ(10, row.GetByte(2));
         ASSERT_EQ(1556, row.GetShort(3));
         ASSERT_EQ(-2556, row.GetShort(4));
@@ -165,7 +167,7 @@ TEST(BinaryRowPartitionComputerTest, TestToAndFromBinaryRow) {
         ASSERT_OK_AND_ASSIGN(BinaryRow row, computer->ToBinaryRow(partition_map));
         ASSERT_EQ(12, row.GetFieldCount());
         ASSERT_EQ(true, row.GetBoolean(0));
-        ASSERT_EQ(-20, row.GetByte(1));
+        ASSERT_EQ(-20, static_cast<int8_t>(row.GetByte(1)));
         ASSERT_EQ(10, row.GetByte(2));
         ASSERT_EQ(1556, row.GetShort(3));
         ASSERT_EQ(-2556, row.GetShort(4));
diff --git a/src/paimon/common/utils/crc32c.cpp b/src/paimon/common/utils/crc32c.cpp
index 1cba7449f..bb01a7b1e 100644
--- a/src/paimon/common/utils/crc32c.cpp
+++ b/src/paimon/common/utils/crc32c.cpp
@@ -22,45 +22,8 @@
 
 namespace paimon {
 uint32_t CRC32C::calculate(const char* data, size_t length, uint32_t crc) {
-#if defined(PAIMON_HAVE_SSE4_2)
-    return crc32c_hw(data, length, crc);
-#else
+    // arrow::internal::crc32 is the zlib polynomial, which its own header spells
+    // out as "different from CRC32C". That is the one this class must produce.
     return arrow::internal::crc32(crc, data, length);
-#endif
 }
-
-#if defined(PAIMON_HAVE_SSE4_2)
-uint32_t CRC32C::crc32c_hw(const char* data, size_t length, uint32_t crc) {
-    crc = ~crc;
-
-    while (length && (reinterpret_cast<uintptr_t>(data) & 7)) {
-        crc = _mm_crc32_u8(crc, *data++);
-        length--;
-    }
-
-    while (length >= 8) {
-        crc = _mm_crc32_u64(crc, *reinterpret_cast<const uint64_t*>(data));
-        data += 8;
-        length -= 8;
-    }
-
-    while (length >= 4) {
-        crc = _mm_crc32_u32(crc, *reinterpret_cast<const uint32_t*>(data));
-        data += 4;
-        length -= 4;
-    }
-
-    while (length >= 2) {
-        crc = _mm_crc32_u16(crc, *reinterpret_cast<const uint16_t*>(data));
-        data += 2;
-        length -= 2;
-    }
-
-    while (length--) {
-        crc = _mm_crc32_u8(crc, *data++);
-    }
-
-    return ~crc;
-}
-#endif
 }  // namespace paimon
diff --git a/src/paimon/common/utils/crc32c.h b/src/paimon/common/utils/crc32c.h
index 7c4db7501..ea0306f56 100644
--- a/src/paimon/common/utils/crc32c.h
+++ b/src/paimon/common/utils/crc32c.h
@@ -20,27 +20,26 @@
 
 #include <cstddef>
 #include <cstdint>
-#if defined(PAIMON_HAVE_SSE4_2)
-#include <nmmintrin.h>
-#endif
+
 #include "paimon/visibility.h"
 
 namespace paimon {
 
-/// CRC32C
+/// Despite the name, this computes CRC-32 with the zlib polynomial 0x04C11DB7,
+/// not Castagnoli CRC-32C. The value is persisted in the SST block trailer and in
+/// the B-tree global index, so it must stay identical across compilers,
+/// architectures and build flags.
+///
+/// An SSE4.2 `_mm_crc32` kernel used to live here behind PAIMON_HAVE_SSE4_2. It
+/// computed Castagnoli, i.e. a different checksum for the same bytes, so it was
+/// removed rather than left switchable. Any future hardware kernel has to
+/// implement the zlib polynomial and be checked against crc32c_test.cpp.
 class PAIMON_EXPORT CRC32C {
  public:
+    /// @param data bytes to checksum
+    /// @param length number of bytes
+    /// @param crc running value to continue from, 0 to start a new checksum
+    /// @return the checksum of `data` appended to the stream `crc` stands for
     static uint32_t calculate(const char* data, size_t length, uint32_t crc = 0);
-
- private:
-#if defined(PAIMON_HAVE_SSE4_2)
-    /// Simd implementation for crc32c.
-    ///
-    /// @param data data to be calculated
-    /// @param length length of data
-    /// @param crc initial crc value
-    /// @return crc32c value
-    static uint32_t crc32c_hw(const char* data, size_t length, uint32_t crc);
-#endif
 };
 }  // namespace paimon
diff --git a/src/paimon/common/utils/crc32c_test.cpp b/src/paimon/common/utils/crc32c_test.cpp
index 10dbbbfeb..9b4533c70 100644
--- a/src/paimon/common/utils/crc32c_test.cpp
+++ b/src/paimon/common/utils/crc32c_test.cpp
@@ -18,20 +18,38 @@
 
 #include "paimon/common/utils/crc32c.h"
 
+#include <array>
+#include <cstddef>
+#include <cstdint>
 #include <cstring>
-#include <limits>
-#include <random>
-#include <set>
-#include <utility>
+#include <string>
 #include <vector>
 
-#include "arrow/util/crc32.h"
 #include "gtest/gtest.h"
-#include "paimon/memory/bytes.h"
-#include "paimon/memory/memory_pool.h"
-#include "paimon/testing/utils/testharness.h"
+
 namespace paimon::test {
 
+namespace {
+
+constexpr size_t kPatternLength = 64;
+constexpr uint32_t kPatternCrc32 = 2227724424u;
+
+/// Built as exact bytes: converting 128-255 to a plain `char` is implementation
+/// defined, and whether `char` is signed depends on the ABI and compiler flags.
+std::vector<uint8_t> MakePattern(size_t length) {
+    std::vector<uint8_t> pattern(length);
+    for (size_t i = 0; i < length; ++i) {
+        pattern[i] = static_cast<uint8_t>((i * 31 + 7) % 256);
+    }
+    return pattern;
+}
+
+const char* AsChars(const std::vector<uint8_t>& bytes) {
+    return reinterpret_cast<const char*>(bytes.data());
+}
+
+}  // namespace
+
 TEST(CRC32CTest, TestSimple) {
     char a = 'a';
     ASSERT_EQ(CRC32C::calculate(&a, 1), 3904355907);
@@ -40,4 +58,72 @@ TEST(CRC32CTest, TestSimple) {
     ASSERT_EQ(CRC32C::calculate(data.c_str(), data.size()), 1311805437);
 }
 
+// The SST block trailer and the B-tree global index persist whatever this
+// produces, so it has to keep matching zlib's CRC-32 (polynomial 0x04C11DB7)
+// byte for byte -- the class name notwithstanding, it is not Castagnoli.
+// Expected values come from Python's zlib.crc32, not from this implementation.
+TEST(CRC32CTest, TestMatchesZlibCrc32ForAllLengthResidues) {
+    struct Case {
+        size_t length;
+        uint32_t crc;
+    };
+    // Lengths cover every residue modulo 8, so each combination of the 8/4/2/1-byte
+    // steps a hardware kernel would take is exercised at least once.
+    const std::vector<Case> cases = {
+        {0, 0u},           {1, 1281784366u},  {2, 3700752837u},
+        {3, 1063713196u},  {4, 1199752182u},  {5, 4062898429u},
+        {6, 690260224u},   {7, 1044920610u},  {8, 2807432232u},
+        {15, 2407004859u}, {16, 104245397u},  {31, 1171692296u},
+        {32, 1227094950u}, {63, 2034968221u}, {kPatternLength, kPatternCrc32},
+    };
+
+    const std::vector<uint8_t> pattern = MakePattern(kPatternLength);
+    for (const Case& test_case : cases) {
+        ASSERT_EQ(CRC32C::calculate(AsChars(pattern), test_case.length), test_case.crc)
+            << "length=" << test_case.length;
+    }
+}
+
+// A hardware kernel would consume leading bytes one at a time until the pointer
+// is aligned, so pinning alignment independence now keeps such a kernel from
+// landing with an alignment-dependent checksum.
+TEST(CRC32CTest, TestIsIndependentOfInputAlignment) {
+    const std::vector<uint8_t> pattern = MakePattern(kPatternLength);
+    // alignas so that `offset` really is the address residue: a char array's own
+    // alignment is 1, so its base could otherwise sit at any residue itself.
+    alignas(8) std::array<char, kPatternLength + 8> shifted{};
+    for (size_t offset = 0; offset < 8; ++offset) {
+        std::memcpy(shifted.data() + offset, pattern.data(), pattern.size());
+        ASSERT_EQ(CRC32C::calculate(shifted.data() + offset, pattern.size()), kPatternCrc32)
+            << "offset=" << offset;
+    }
+}
+
+// SstFileWriter checksums a block and then folds the compression byte into the
+// running value, so chaining must agree with a single pass over the same bytes.
+TEST(CRC32CTest, TestRunningChecksumMatchesSinglePass) {
+    std::string data = "hello paimon c++";
+    for (size_t split = 0; split <= data.size(); ++split) {
+        uint32_t running = CRC32C::calculate(data.c_str(), split);
+        running = CRC32C::calculate(data.c_str() + split, data.size() - split, running);
+        ASSERT_EQ(running, CRC32C::calculate(data.c_str(), data.size())) << "split=" << split;
+    }
+}
+
+// The test above only shows the two paths agree with each other, which would
+// still hold if both were wrong. Pin the seeded path to zlib's own output too.
+// Expected values come from Python's zlib.crc32(data, seed).
+TEST(CRC32CTest, TestSeededChecksumMatchesZlibCrc32) {
+    constexpr uint32_t kSeed = 0x12345678u;
+    const std::vector<uint8_t> pattern = MakePattern(kPatternLength);
+
+    // Continuing a stream without appending anything leaves the running value alone.
+    ASSERT_EQ(CRC32C::calculate(AsChars(pattern), 0, kSeed), kSeed);
+
+    ASSERT_EQ(CRC32C::calculate(AsChars(pattern), pattern.size(), kSeed), 3207811452u);
+
+    std::string data = "hello paimon c++";
+    ASSERT_EQ(CRC32C::calculate(data.c_str(), data.size(), kSeed), 671566677u);
+}
+
 }  // namespace paimon::test
diff --git a/src/paimon/common/utils/data_converter_utils_test.cpp b/src/paimon/common/utils/data_converter_utils_test.cpp
index b0078552c..75913eccd 100644
--- a/src/paimon/common/utils/data_converter_utils_test.cpp
+++ b/src/paimon/common/utils/data_converter_utils_test.cpp
@@ -20,6 +20,7 @@
 #include "paimon/common/utils/data_converter_utils.h"
 
 #include <cstddef>
+#include <cstdint>
 #include <memory>
 #include <vector>
 
@@ -71,7 +72,9 @@ TEST(DataConverterUtilsTest, TestDataToBinaryRowConverterWithLegacyPartitionName
     ASSERT_EQ(data.size(), row.GetFieldCount());
     ASSERT_EQ(true, row.GetBoolean(0));
     ASSERT_EQ(10, row.GetByte(1));
-    ASSERT_EQ(-20, row.GetByte(2));
+    // GetByte() returns a plain char, whose signedness the target ABI decides and
+    // -fsigned-char / -funsigned-char can flip; the AArch64 Linux ABI makes it unsigned.
+    ASSERT_EQ(-20, static_cast<int8_t>(row.GetByte(2)));
     ASSERT_EQ(1556, row.GetShort(3));
     ASSERT_EQ(-2556, row.GetShort(4));
     ASSERT_EQ(348489, row.GetInt(5));
@@ -125,7 +128,9 @@ TEST(DataConverterUtilsTest, TestDataToBinaryRowConverterWithNoLegacyPartitionNa
     ASSERT_EQ(data.size(), row.GetFieldCount());
     ASSERT_EQ(true, row.GetBoolean(0));
     ASSERT_EQ(10, row.GetByte(1));
-    ASSERT_EQ(-20, row.GetByte(2));
+    // GetByte() returns a plain char, whose signedness the target ABI decides and
+    // -fsigned-char / -funsigned-char can flip; the AArch64 Linux ABI makes it unsigned.
+    ASSERT_EQ(-20, static_cast<int8_t>(row.GetByte(2)));
     ASSERT_EQ(1556, row.GetShort(3));
     ASSERT_EQ(-2556, row.GetShort(4));
     ASSERT_EQ(348489, row.GetInt(5));
diff --git a/src/paimon/core/casting/cast_executor_test.cpp b/src/paimon/core/casting/cast_executor_test.cpp
index ed540ae49..3905ba04d 100644
--- a/src/paimon/core/casting/cast_executor_test.cpp
+++ b/src/paimon/core/casting/cast_executor_test.cpp
@@ -134,6 +134,14 @@ class CastExecutorTest : public ::testing::Test {
         return target_array.status().ToString();
     }
 
+    /// Casts through the fixture's memory pool, which the test bodies cannot reach themselves.
+    Result<std::shared_ptr<arrow::Array>> CastArray(
+        const std::shared_ptr<CastExecutor>& cast_executor,
+        const std::shared_ptr<arrow::Array>& src_array,
+        const std::shared_ptr<arrow::DataType>& target_type) const {
+        return cast_executor->Cast(src_array, target_type, arrow_pool_.get());
+    }
+
     template <typename Type>
     Literal CreateLiteral(const FieldType& type, const Type& data) const {
         if constexpr (std::is_same_v<Type, std::string>) {
@@ -154,6 +162,9 @@ class CastExecutorTest : public ::testing::Test {
                             const FieldType& src_type, const std::vector<SrcType>& src_data,
                             const std::shared_ptr<arrow::DataType>& target_type,
                             const std::vector<TargetType>& target_data) const {
+        // Without this, a target vector longer than the source is silently ignored and a shorter
+        // one reads out of bounds.
+        ASSERT_EQ(src_data.size(), target_data.size());
         FieldType target_field_type = FieldTypeUtils::ConvertToFieldType(target_type->id()).value();
         std::vector<Literal> src_literals;
         std::vector<Literal> target_literals;
@@ -215,6 +226,8 @@ class CastExecutorTest : public ::testing::Test {
     static constexpr int64_t NANOS_PER_SECOND = 1000000000l;
     static constexpr int64_t MILLIS_PER_SECOND = 1000l;
     static constexpr int64_t NANOS_PER_MILLIS = 1000000l;
+    static constexpr int32_t MAX_INT32 = std::numeric_limits<int32_t>::max();
+    static constexpr int32_t MIN_INT32 = std::numeric_limits<int32_t>::min();
     static constexpr int64_t MAX_INT64 = std::numeric_limits<int64_t>::max();
     static constexpr int64_t MIN_INT64 = std::numeric_limits<int64_t>::min();
     static constexpr float MAX_FLOAT = std::numeric_limits<float>::max();
@@ -380,42 +393,58 @@ TEST_F(CastExecutorTest, TestNumericPrimitiveCastExecutorCastLiteral) {
     }
     // test src type FLOAT
     {
-        // Java Paimon cast MAX_FLOAT to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
-        std::vector<int8_t> target_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<float> src_data = {1.1,       2.2,      3.3,       -10.01, 0,
+                                       127.9,     -128.9,   300.9,     -300.9, MAX_FLOAT,
+                                       MIN_FLOAT, INFINITY, -INFINITY, NAN};
+        std::vector<int8_t> target_data = {1, 2, 3, -10, 0, 127, -128, 44, -44, -1, 0, -1, 0, 0};
         CheckLiteralResult<float, int8_t>(cast_executor, FieldType::FLOAT, src_data, arrow::int8(),
                                           target_data);
     }
     {
-        // Java Paimon cast MAX_FLOAT to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
-        std::vector<int16_t> target_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<float> src_data = {1.1,      2.2,       3.3,     -10.01,    0,
+                                       32767.9,  -32768.9,  40000.9, MAX_FLOAT, MIN_FLOAT,
+                                       INFINITY, -INFINITY, NAN};
+        std::vector<int16_t> target_data = {1,      2,  3, -10, 0, 32767, -32768,
+                                            -25536, -1, 0, -1,  0, 0};
         CheckLiteralResult<float, int16_t>(cast_executor, FieldType::FLOAT, src_data,
                                            arrow::int16(), target_data);
     }
     {
-        // Java Paimon cast MAX_FLOAT to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast INFINITY to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to -2147483648
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
+        const float max_float_convertible_to_int32 = std::nextafter(2147483648.0f, 0.0f);
+        std::vector<float> src_data = {1.1,
+                                       2.2,
+                                       3.3,
+                                       -10.01,
+                                       0,
+                                       -2147483648.0f,
+                                       max_float_convertible_to_int32,
+                                       MAX_FLOAT,
+                                       MIN_FLOAT,
+                                       INFINITY,
+                                       -INFINITY,
+                                       NAN};
         std::vector<int32_t> target_data = {
-            1, 2, 3, -10, 0, -2147483648, -2147483648, -2147483648, -2147483648, -2147483648};
+            1, 2, 3, -10, 0, MIN_INT32, 2147483520, MAX_INT32, MIN_INT32, MAX_INT32, MIN_INT32, 0};
         CheckLiteralResult<float, int32_t>(cast_executor, FieldType::FLOAT, src_data,
                                            arrow::int32(), target_data);
     }
     {
-        // Java Paimon cast MAX_FLOAT to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast INFINITY to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to MIN_INT64
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
-        std::vector<int64_t> target_data = {1,         2,         3,         -10,       0,
-                                            MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64};
+        const float max_float_convertible_to_int64 = std::nextafter(9223372036854775808.0f, 0.0f);
+        std::vector<float> src_data = {1.1,
+                                       2.2,
+                                       3.3,
+                                       -10.01,
+                                       0,
+                                       static_cast<float>(MIN_INT64),
+                                       max_float_convertible_to_int64,
+                                       MAX_FLOAT,
+                                       MIN_FLOAT,
+                                       INFINITY,
+                                       -INFINITY,
+                                       NAN};
+        std::vector<int64_t> target_data = {
+            1,         2,         3,         -10,       0, MIN_INT64, 9223371487098961920,
+            MAX_INT64, MIN_INT64, MAX_INT64, MIN_INT64, 0};
         CheckLiteralResult<float, int64_t>(cast_executor, FieldType::FLOAT, src_data,
                                            arrow::int64(), target_data);
     }
@@ -445,42 +474,48 @@ TEST_F(CastExecutorTest, TestNumericPrimitiveCastExecutorCastLiteral) {
     }
     // test src type DOUBLE
     {
-        // Java Paimon cast MAX_DOUBLE to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
-        std::vector<int8_t> target_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<double> src_data = {1.1,        2.2,      3.3,       -10.01, 0,
+                                        127.9,      -128.9,   300.9,     -300.9, MAX_DOUBLE,
+                                        MIN_DOUBLE, INFINITY, -INFINITY, NAN};
+        std::vector<int8_t> target_data = {1, 2, 3, -10, 0, 127, -128, 44, -44, -1, 0, -1, 0, 0};
         CheckLiteralResult<double, int8_t>(cast_executor, FieldType::DOUBLE, src_data,
                                            arrow::int8(), target_data);
     }
     {
-        // Java Paimon cast MAX_DOUBLE to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
-        std::vector<int16_t> target_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<double> src_data = {1.1,      2.2,       3.3,     -10.01,     0,
+                                        32767.9,  -32768.9,  40000.9, MAX_DOUBLE, MIN_DOUBLE,
+                                        INFINITY, -INFINITY, NAN};
+        std::vector<int16_t> target_data = {1,      2,  3, -10, 0, 32767, -32768,
+                                            -25536, -1, 0, -1,  0, 0};
         CheckLiteralResult<double, int16_t>(cast_executor, FieldType::DOUBLE, src_data,
                                             arrow::int16(), target_data);
     }
     {
-        // Java Paimon cast MAX_DOUBLE to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast INFINITY to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to -2147483648
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
+        std::vector<double> src_data = {1.1,        2.2,          3.3,           -10.01,
+                                        0,          2147483647.9, -2147483648.9, MAX_DOUBLE,
+                                        MIN_DOUBLE, INFINITY,     -INFINITY,     NAN};
         std::vector<int32_t> target_data = {
-            1, 2, 3, -10, 0, -2147483648, -2147483648, -2147483648, -2147483648, -2147483648};
+            1, 2, 3, -10, 0, MAX_INT32, MIN_INT32, MAX_INT32, MIN_INT32, MAX_INT32, MIN_INT32, 0};
         CheckLiteralResult<double, int32_t>(cast_executor, FieldType::DOUBLE, src_data,
                                             arrow::int32(), target_data);
     }
     {
-        // Java Paimon cast MAX_DOUBLE to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast INFINITY to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to MIN_INT64
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
-        std::vector<int64_t> target_data = {1,         2,         3,         -10,       0,
-                                            MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64};
+        const double max_double_convertible_to_int64 = std::nextafter(9223372036854775808.0, 0.0);
+        std::vector<double> src_data = {1.1,
+                                        2.2,
+                                        3.3,
+                                        -10.01,
+                                        0,
+                                        static_cast<double>(MIN_INT64),
+                                        max_double_convertible_to_int64,
+                                        MAX_DOUBLE,
+                                        MIN_DOUBLE,
+                                        INFINITY,
+                                        -INFINITY,
+                                        NAN};
+        std::vector<int64_t> target_data = {
+            1,         2,         3,         -10,       0, MIN_INT64, 9223372036854774784,
+            MAX_INT64, MIN_INT64, MAX_INT64, MIN_INT64, 0};
         CheckLiteralResult<double, int64_t>(cast_executor, FieldType::DOUBLE, src_data,
                                             arrow::int64(), target_data);
     }
@@ -590,51 +625,65 @@ TEST_F(CastExecutorTest, TestNumericPrimitiveCastExecutorCastArray) {
     }
     // test src type float
     {
-        // Java Paimon cast MAX_FLOAT to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
+        std::vector<float> src_data = {1.1,       2.2,      3.3,       -10.01, 0,
+                                       127.9,     -128.9,   300.9,     -300.9, MAX_FLOAT,
+                                       MIN_FLOAT, INFINITY, -INFINITY, NAN};
         auto src_array = MakeArrowArray<float, arrow::FloatBuilder>(arrow::float32(), src_data);
-        std::vector<int8_t> expected_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<int8_t> expected_data = {1, 2, 3, -10, 0, 127, -128, 44, -44, -1, 0, -1, 0, 0};
         auto expected_array =
             MakeArrowArray<int8_t, arrow::Int8Builder>(arrow::int8(), expected_data);
         CheckArrayResult(cast_executor, arrow::int8(), src_array, expected_array);
     }
     {
-#ifndef NDEBUG
-        // Java Paimon cast MAX_FLOAT to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
+        std::vector<float> src_data = {1.1,      2.2,       3.3,     -10.01,    0,
+                                       32767.9,  -32768.9,  40000.9, MAX_FLOAT, MIN_FLOAT,
+                                       INFINITY, -INFINITY, NAN};
         auto src_array = MakeArrowArray<float, arrow::FloatBuilder>(arrow::float32(), src_data);
-        std::vector<int16_t> expected_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<int16_t> expected_data = {1,      2,  3, -10, 0, 32767, -32768,
+                                              -25536, -1, 0, -1,  0, 0};
         auto expected_array =
             MakeArrowArray<int16_t, arrow::Int16Builder>(arrow::int16(), expected_data);
         CheckArrayResult(cast_executor, arrow::int16(), src_array, expected_array);
-#endif
     }
     {
-        // Java Paimon cast MAX_FLOAT to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast INFINITY to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to -2147483648
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
+        const float max_float_convertible_to_int32 = std::nextafter(2147483648.0f, 0.0f);
+        std::vector<float> src_data = {1.1,
+                                       2.2,
+                                       3.3,
+                                       -10.01,
+                                       0,
+                                       -2147483648.0f,
+                                       max_float_convertible_to_int32,
+                                       MAX_FLOAT,
+                                       MIN_FLOAT,
+                                       INFINITY,
+                                       -INFINITY,
+                                       NAN};
         auto src_array = MakeArrowArray<float, arrow::FloatBuilder>(arrow::float32(), src_data);
         std::vector<int32_t> expected_data = {
-            1, 2, 3, -10, 0, -2147483648, -2147483648, -2147483648, -2147483648, -2147483648};
+            1, 2, 3, -10, 0, MIN_INT32, 2147483520, MAX_INT32, MIN_INT32, MAX_INT32, MIN_INT32, 0};
         auto expected_array =
             MakeArrowArray<int32_t, arrow::Int32Builder>(arrow::int32(), expected_data);
         CheckArrayResult(cast_executor, arrow::int32(), src_array, expected_array);
     }
     {
-        // Java Paimon cast MAX_FLOAT to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast INFINITY to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to MIN_INT64
-        std::vector<float> src_data = {1.1,       2.2,       3.3,      -10.01,    0,
-                                       MAX_FLOAT, MIN_FLOAT, INFINITY, -INFINITY, NAN};
+        const float max_float_convertible_to_int64 = std::nextafter(9223372036854775808.0f, 0.0f);
+        std::vector<float> src_data = {1.1,
+                                       2.2,
+                                       3.3,
+                                       -10.01,
+                                       0,
+                                       static_cast<float>(MIN_INT64),
+                                       max_float_convertible_to_int64,
+                                       MAX_FLOAT,
+                                       MIN_FLOAT,
+                                       INFINITY,
+                                       -INFINITY,
+                                       NAN};
         auto src_array = MakeArrowArray<float, arrow::FloatBuilder>(arrow::float32(), src_data);
         std::vector<int64_t> expected_data = {
-            1, 2, 3, -10, 0, MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64};
+            1,         2,         3,         -10,       0, MIN_INT64, 9223371487098961920,
+            MAX_INT64, MIN_INT64, MAX_INT64, MIN_INT64, 0};
         auto expected_array =
             MakeArrowArray<int64_t, arrow::Int64Builder>(arrow::int64(), expected_data);
         CheckArrayResult(cast_executor, arrow::int64(), src_array, expected_array);
@@ -662,49 +711,55 @@ TEST_F(CastExecutorTest, TestNumericPrimitiveCastExecutorCastArray) {
 
     // test src type double
     {
-        // Java Paimon cast MAX_DOUBLE to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
+        std::vector<double> src_data = {1.1,        2.2,      3.3,       -10.01, 0,
+                                        127.9,      -128.9,   300.9,     -300.9, MAX_DOUBLE,
+                                        MIN_DOUBLE, INFINITY, -INFINITY, NAN};
         auto src_array = MakeArrowArray<double, arrow::DoubleBuilder>(arrow::float64(), src_data);
-        std::vector<int8_t> expected_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<int8_t> expected_data = {1, 2, 3, -10, 0, 127, -128, 44, -44, -1, 0, -1, 0, 0};
         auto expected_array =
             MakeArrowArray<int8_t, arrow::Int8Builder>(arrow::int8(), expected_data);
         CheckArrayResult(cast_executor, arrow::int8(), src_array, expected_array);
     }
     {
-        // Java Paimon cast MAX_DOUBLE to -1 while C++ Paimon cast to 0
-        // Java Paimon cast INFINITY to -1 while C++ Paimon cast to 0
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
+        std::vector<double> src_data = {1.1,      2.2,       3.3,     -10.01,     0,
+                                        32767.9,  -32768.9,  40000.9, MAX_DOUBLE, MIN_DOUBLE,
+                                        INFINITY, -INFINITY, NAN};
         auto src_array = MakeArrowArray<double, arrow::DoubleBuilder>(arrow::float64(), src_data);
-        std::vector<int16_t> expected_data = {1, 2, 3, -10, 0, 0, 0, 0, 0, 0};
+        std::vector<int16_t> expected_data = {1,      2,  3, -10, 0, 32767, -32768,
+                                              -25536, -1, 0, -1,  0, 0};
         auto expected_array =
             MakeArrowArray<int16_t, arrow::Int16Builder>(arrow::int16(), expected_data);
         CheckArrayResult(cast_executor, arrow::int16(), src_array, expected_array);
     }
     {
-        // Java Paimon cast MAX_DOUBLE to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast INFINITY to 2147483647 while C++ Paimon cast to -2147483648
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to -2147483648
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
+        std::vector<double> src_data = {1.1,        2.2,          3.3,           -10.01,
+                                        0,          2147483647.9, -2147483648.9, MAX_DOUBLE,
+                                        MIN_DOUBLE, INFINITY,     -INFINITY,     NAN};
         auto src_array = MakeArrowArray<double, arrow::DoubleBuilder>(arrow::float64(), src_data);
         std::vector<int32_t> expected_data = {
-            1, 2, 3, -10, 0, -2147483648, -2147483648, -2147483648, -2147483648, -2147483648};
+            1, 2, 3, -10, 0, MAX_INT32, MIN_INT32, MAX_INT32, MIN_INT32, MAX_INT32, MIN_INT32, 0};
         auto expected_array =
             MakeArrowArray<int32_t, arrow::Int32Builder>(arrow::int32(), expected_data);
         CheckArrayResult(cast_executor, arrow::int32(), src_array, expected_array);
     }
     {
-        // Java Paimon cast MAX_DOUBLE to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast INFINITY to MAX_INT64 while C++ Paimon cast to MIN_INT64
-        // Java Paimon cast NAN to 0 while C++ Paimon cast to MIN_INT64
-        std::vector<double> src_data = {1.1,        2.2,        3.3,      -10.01,    0,
-                                        MAX_DOUBLE, MIN_DOUBLE, INFINITY, -INFINITY, NAN};
+        const double max_double_convertible_to_int64 = std::nextafter(9223372036854775808.0, 0.0);
+        std::vector<double> src_data = {1.1,
+                                        2.2,
+                                        3.3,
+                                        -10.01,
+                                        0,
+                                        static_cast<double>(MIN_INT64),
+                                        max_double_convertible_to_int64,
+                                        MAX_DOUBLE,
+                                        MIN_DOUBLE,
+                                        INFINITY,
+                                        -INFINITY,
+                                        NAN};
         auto src_array = MakeArrowArray<double, arrow::DoubleBuilder>(arrow::float64(), src_data);
         std::vector<int64_t> expected_data = {
-            1, 2, 3, -10, 0, MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64, MIN_INT64};
+            1,         2,         3,         -10,       0, MIN_INT64, 9223372036854774784,
+            MAX_INT64, MIN_INT64, MAX_INT64, MIN_INT64, 0};
         auto expected_array =
             MakeArrowArray<int64_t, arrow::Int64Builder>(arrow::int64(), expected_data);
         CheckArrayResult(cast_executor, arrow::int64(), src_array, expected_array);
@@ -731,6 +786,50 @@ TEST_F(CastExecutorTest, TestNumericPrimitiveCastExecutorCastArray) {
     }
 }
 
+TEST_F(CastExecutorTest, TestNumericPrimitiveCastExecutorFloatingToIntegerArrayEdgeCases) {
+    auto cast_executor = std::make_shared<NumericPrimitiveCastExecutor>();
+    // float and double to integer no longer go through Arrow's kernel, so the edge cases of the
+    // conversion this executor does itself are covered here.
+    std::vector<float> src_data = {1.1, MAX_FLOAT, -10.01, NAN, 127.9};
+    auto src_array = MakeArrowArray<float, arrow::FloatBuilder>(arrow::float32(), src_data);
+    std::vector<int8_t> expected_data = {1, -1, -10, 0, 127};
+    auto expected_array = MakeArrowArray<int8_t, arrow::Int8Builder>(arrow::int8(), expected_data);
+    {
+        // MakeArrowArray() appends a trailing null, so this also covers null passthrough.
+        CheckArrayResult(cast_executor, arrow::int8(), src_array, expected_array);
+    }
+    {
+        // An empty array must not reserve or append anything.
+        auto empty_src_array = src_array->Slice(0, 0);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> target_array,
+                             CastArray(cast_executor, empty_src_array, arrow::int8()));
+        ASSERT_EQ(0, target_array->length());
+        ASSERT_TRUE(target_array->type()->Equals(arrow::int8()));
+    }
+    {
+        // A slice has a non-zero offset, which the conversion has to honour for both the values
+        // and the validity bitmap.
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> target_array,
+                             CastArray(cast_executor, src_array->Slice(2), arrow::int8()));
+        ASSERT_TRUE(target_array->Equals(expected_array->Slice(2),
+                                         arrow::EqualOptions::Defaults().nans_equal(true)))
+            << "target:" << target_array->ToString()
+            << "expected:" << expected_array->Slice(2)->ToString();
+    }
+    {
+        // The double instantiation shares the template, but pin its slice handling too.
+        std::vector<double> double_src_data = {1.1, MAX_DOUBLE, -10.01, NAN, 127.9};
+        auto double_src_array =
+            MakeArrowArray<double, arrow::DoubleBuilder>(arrow::float64(), double_src_data);
+        ASSERT_OK_AND_ASSIGN(std::shared_ptr<arrow::Array> target_array,
+                             CastArray(cast_executor, double_src_array->Slice(2), arrow::int8()));
+        ASSERT_TRUE(target_array->Equals(expected_array->Slice(2),
+                                         arrow::EqualOptions::Defaults().nans_equal(true)))
+            << "target:" << target_array->ToString()
+            << "expected:" << expected_array->Slice(2)->ToString();
+    }
+}
+
 TEST_F(CastExecutorTest, TestBooleanToNumericCastExecutorCastLiteral) {
     auto cast_executor = std::make_shared<BooleanToNumericCastExecutor>();
     std::vector<bool> src_data = {true, false};
diff --git a/src/paimon/core/casting/numeric_primitive_cast_executor.cpp b/src/paimon/core/casting/numeric_primitive_cast_executor.cpp
index 785189e18..3963220fd 100644
--- a/src/paimon/core/casting/numeric_primitive_cast_executor.cpp
+++ b/src/paimon/core/casting/numeric_primitive_cast_executor.cpp
@@ -20,11 +20,14 @@
 #include "paimon/core/casting/numeric_primitive_cast_executor.h"
 
 #include <cstdint>
+#include <memory>
 #include <string>
 
+#include "arrow/api.h"
 #include "arrow/compute/cast.h"
 #include "arrow/type.h"
 #include "fmt/format.h"
+#include "paimon/common/utils/arrow/status_utils.h"
 #include "paimon/common/utils/field_type_utils.h"
 #include "paimon/core/casting/casting_utils.h"
 #include "paimon/defs.h"
@@ -204,9 +207,76 @@ Result<Literal> NumericPrimitiveCastExecutor::Cast(
     return iter->second(literal);
 }
 
+namespace {
+template <typename SrcArrayType, typename TargetBuilderType, typename TargetType>
+Result<std::shared_ptr<arrow::Array>> JavaCastFloatingArray(const arrow::Array& src_array,
+                                                            arrow::MemoryPool* pool) {
+    const auto& typed_src_array = static_cast<const SrcArrayType&>(src_array);
+    TargetBuilderType builder(pool);
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Reserve(typed_src_array.length()));
+    for (int64_t i = 0; i < typed_src_array.length(); i++) {
+        if (typed_src_array.IsNull(i)) {
+            builder.UnsafeAppendNull();
+        } else {
+            builder.UnsafeAppend(
+                NumericPrimitiveCastExecutor::JavaFloatingToIntegerCast<TargetType>(
+                    typed_src_array.Value(i)));
+        }
+    }
+    std::shared_ptr<arrow::Array> target_array;
+    PAIMON_RETURN_NOT_OK_FROM_ARROW(builder.Finish(&target_array));
+    return target_array;
+}
+
+template <typename SrcArrayType>
+Result<std::shared_ptr<arrow::Array>> JavaCastFloatingArrayToInteger(
+    const arrow::Array& src_array, const std::shared_ptr<arrow::DataType>& target_type,
+    arrow::MemoryPool* pool) {
+    switch (target_type->id()) {
+        case arrow::Type::INT8:
+            return JavaCastFloatingArray<SrcArrayType, arrow::Int8Builder, int8_t>(src_array, pool);
+        case arrow::Type::INT16:
+            return JavaCastFloatingArray<SrcArrayType, arrow::Int16Builder, int16_t>(src_array,
+                                                                                     pool);
+        case arrow::Type::INT32:
+            return JavaCastFloatingArray<SrcArrayType, arrow::Int32Builder, int32_t>(src_array,
+                                                                                     pool);
+        case arrow::Type::INT64:
+            return JavaCastFloatingArray<SrcArrayType, arrow::Int64Builder, int64_t>(src_array,
+                                                                                     pool);
+        default:
+            return Status::Invalid(
+                fmt::format("cast array in NumericPrimitiveCastExecutor failed: {} is not an "
+                            "integer target type",
+                            target_type->ToString()));
+    }
+}
+
+// Deliberately not arrow::is_integer(): that also matches the unsigned types, which
+// JavaCastFloatingArrayToInteger() does not handle and which still belong to Arrow's kernel.
+bool IsSignedIntegerType(arrow::Type::type type) {
+    return type == arrow::Type::INT8 || type == arrow::Type::INT16 || type == arrow::Type::INT32 ||
+           type == arrow::Type::INT64;
+}
+}  // namespace
+
 Result<std::shared_ptr<arrow::Array>> NumericPrimitiveCastExecutor::Cast(
     const std::shared_ptr<arrow::Array>& array, const std::shared_ptr<arrow::DataType>& target_type,
     arrow::MemoryPool* pool) const {
+    // Arrow's floating point to integer kernel is a plain static_cast, annotated
+    // ARROW_DISABLE_UBSAN("float-cast-overflow"), so it is undefined for a value that does not
+    // fit and answers it differently per architecture. Convert those here instead, with the same
+    // Java semantics the literal overload uses, so that stats converted through one path and
+    // column data converted through the other agree. This trades Arrow's vectorized kernel for a
+    // scalar loop, which these conversions, reached from schema evolution, can afford.
+    if (IsSignedIntegerType(target_type->id())) {
+        if (array->type_id() == arrow::Type::FLOAT) {
+            return JavaCastFloatingArrayToInteger<arrow::FloatArray>(*array, target_type, pool);
+        }
+        if (array->type_id() == arrow::Type::DOUBLE) {
+            return JavaCastFloatingArrayToInteger<arrow::DoubleArray>(*array, target_type, pool);
+        }
+    }
     arrow::compute::CastOptions options = arrow::compute::CastOptions::Safe();
     options.allow_int_overflow = true;
     options.allow_float_truncate = true;
diff --git a/src/paimon/core/casting/numeric_primitive_cast_executor.h b/src/paimon/core/casting/numeric_primitive_cast_executor.h
index d84cd9f52..c6e2d5b15 100644
--- a/src/paimon/core/casting/numeric_primitive_cast_executor.h
+++ b/src/paimon/core/casting/numeric_primitive_cast_executor.h
@@ -18,9 +18,13 @@
  */
 
 #pragma once
+#include <cmath>
+#include <cstdint>
 #include <functional>
+#include <limits>
 #include <map>
 #include <memory>
+#include <type_traits>
 #include <utility>
 
 #include "arrow/array/array_base.h"
@@ -46,14 +50,71 @@ class NumericPrimitiveCastExecutor : public CastExecutor {
                                                const std::shared_ptr<arrow::DataType>& target_type,
                                                arrow::MemoryPool* pool) const override;
 
+    /// Converts a floating point value to an integer type the way Java does, where a plain
+    /// `static_cast` would be undefined behavior with an architecture dependent result (see
+    /// docs/code-style.md): `NaN` becomes 0, and a value that does not fit saturates.
+    ///
+    /// Java reaches TINYINT and SMALLINT by narrowing an `int`, so those saturate at the int32
+    /// bounds and are then truncated to width, which is why `(byte) Float.MAX_VALUE` is -1 and
+    /// not 127. BIGINT saturates at the int64 bounds.
+    template <typename TargetType, typename SrcType>
+    static TargetType JavaFloatingToIntegerCast(SrcType value) {
+        static_assert(std::is_floating_point_v<SrcType>, "source must be floating point");
+        static_assert(std::is_integral_v<TargetType>, "target must be integral");
+        static_assert(std::is_signed_v<TargetType>, "target must be signed");
+        using WideType = std::conditional_t<std::is_same_v<TargetType, int64_t>, int64_t, int32_t>;
+        if (std::isnan(value)) {
+            return 0;
+        }
+        WideType wide_value;
+        // Comparing against the bounds converted to SrcType is what keeps the truncation defined.
+        // The standard leaves that conversion a choice between the two adjacent representable
+        // values; under the IEEE round-to-nearest the supported toolchains use, an unrepresentable
+        // upper bound rounds up to a power of two, so every value that reaches the truncation is
+        // inside the integer range.
+        if (value >= static_cast<SrcType>(std::numeric_limits<WideType>::max())) {
+            wide_value = std::numeric_limits<WideType>::max();
+        } else if (value <= static_cast<SrcType>(std::numeric_limits<WideType>::lowest())) {
+            wide_value = std::numeric_limits<WideType>::lowest();
+        } else {
+            wide_value = static_cast<WideType>(value);
+        }
+        return NarrowToTwosComplement<TargetType>(wide_value);
+    }
+
  private:
+    /// Truncates to the width of TargetType keeping the low bits, which is what Java narrowing
+    /// does. Converting an out of range value to a signed type directly is implementation
+    /// defined until C++20, so the low bits are taken through the unsigned type, where the
+    /// wrap-around is defined, and only an in-range value is converted back.
+    template <typename TargetType, typename WideType>
+    static TargetType NarrowToTwosComplement(WideType value) {
+        static_assert(std::is_signed_v<TargetType>, "target must be signed");
+        using UnsignedTarget = std::make_unsigned_t<TargetType>;
+        constexpr TargetType kMin = std::numeric_limits<TargetType>::min();
+        // The unsigned value of the minimum is 2^(width-1), the point where the low bits start
+        // standing for a negative number.
+        constexpr auto kHalf = static_cast<UnsignedTarget>(kMin);
+        const auto low_bits = static_cast<UnsignedTarget>(value);
+        if (low_bits < kHalf) {
+            return static_cast<TargetType>(low_bits);
+        }
+        // low_bits - kHalf is in range, so converting it back is defined, and adding the minimum
+        // gives the negative value those bits stand for.
+        return static_cast<TargetType>(static_cast<TargetType>(low_bits - kHalf) + kMin);
+    }
+
     template <typename SrcType, typename TargetType>
     static Literal CastLiteral(const Literal& literal, const FieldType& target_type) {
         if (literal.IsNull()) {
             return Literal(target_type);
         }
         SrcType value = literal.GetValue<SrcType>();
-        return Literal(static_cast<TargetType>(value));
+        if constexpr (std::is_floating_point_v<SrcType> && std::is_integral_v<TargetType>) {
+            return Literal(JavaFloatingToIntegerCast<TargetType>(value));
+        } else {
+            return Literal(static_cast<TargetType>(value));
+        }
     }
 
  private:
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_max_agg.h b/src/paimon/core/mergetree/compact/aggregate/field_max_agg.h
index 6273a5749..f41764e4b 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_max_agg.h
+++ b/src/paimon/core/mergetree/compact/aggregate/field_max_agg.h
@@ -18,6 +18,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <memory>
 #include <string>
 
@@ -60,6 +61,17 @@ class FieldMaxAgg : public FieldAggregator {
         arrow::Type::type type = field_type->id();
         switch (type) {
             case arrow::Type::type::INT8:
+                // The variant holds TINYINT as a plain char, so comparing the variants directly
+                // would follow the ABI's signedness for char and order negative values above
+                // positive ones where it is unsigned. Compare the signed value it stands for.
+                return FieldMaxFunc([](const VariantType& accumulator,
+                                       const VariantType& input_field) -> VariantType {
+                    auto accumulator_value =
+                        static_cast<int8_t>(DataDefine::GetVariantValue<char>(accumulator));
+                    auto input_value =
+                        static_cast<int8_t>(DataDefine::GetVariantValue<char>(input_field));
+                    return accumulator_value < input_value ? input_field : accumulator;
+                });
             case arrow::Type::type::INT16:
             case arrow::Type::type::INT32:
             case arrow::Type::type::DATE32:
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_min_agg.h b/src/paimon/core/mergetree/compact/aggregate/field_min_agg.h
index 198e4cf11..8201a820e 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_min_agg.h
+++ b/src/paimon/core/mergetree/compact/aggregate/field_min_agg.h
@@ -18,6 +18,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <memory>
 #include <string>
 
@@ -60,6 +61,17 @@ class FieldMinAgg : public FieldAggregator {
         arrow::Type::type type = field_type->id();
         switch (type) {
             case arrow::Type::type::INT8:
+                // The variant holds TINYINT as a plain char, so comparing the variants directly
+                // would follow the ABI's signedness for char and order negative values above
+                // positive ones where it is unsigned. Compare the signed value it stands for.
+                return FieldMinFunc([](const VariantType& accumulator,
+                                       const VariantType& input_field) -> VariantType {
+                    auto accumulator_value =
+                        static_cast<int8_t>(DataDefine::GetVariantValue<char>(accumulator));
+                    auto input_value =
+                        static_cast<int8_t>(DataDefine::GetVariantValue<char>(input_field));
+                    return accumulator_value < input_value ? accumulator : input_field;
+                });
             case arrow::Type::type::INT16:
             case arrow::Type::type::INT32:
             case arrow::Type::type::DATE32:
diff --git a/src/paimon/core/mergetree/compact/aggregate/field_min_max_agg_test.cpp b/src/paimon/core/mergetree/compact/aggregate/field_min_max_agg_test.cpp
index c5dd1f337..e58df125a 100644
--- a/src/paimon/core/mergetree/compact/aggregate/field_min_max_agg_test.cpp
+++ b/src/paimon/core/mergetree/compact/aggregate/field_min_max_agg_test.cpp
@@ -147,6 +147,9 @@ TEST(FieldMinMaxAggTest, TestVariantType) {
     };
 
     CheckResult(arrow::int8(), static_cast<char>(100), static_cast<char>(15));
+    // A negative TINYINT: the variant holds it as a plain char, so where the ABI makes char
+    // unsigned, comparing the variants directly would order -20 above 10.
+    CheckResult(arrow::int8(), static_cast<char>(10), static_cast<char>(-20));
     CheckResult(arrow::int16(), static_cast<int16_t>(100), static_cast<int16_t>(15));
     CheckResult(arrow::int32(), static_cast<int32_t>(100), static_cast<int32_t>(15));
     CheckResult(arrow::date32(), static_cast<int32_t>(100), static_cast<int32_t>(15));