diff --git a/be/src/core/column/column_varbinary.cpp b/be/src/core/column/column_varbinary.cpp
index 4a54ca421d0a65..71dd2f91e6b07f 100644
--- a/be/src/core/column/column_varbinary.cpp
+++ b/be/src/core/column/column_varbinary.cpp
@@ -31,6 +31,26 @@
 #include "exec/sort/sort_block.h"
 
 namespace doris {
+
+void ColumnVarbinary::insert_many_continuous_binary_data(const char* data, const uint32_t* offsets,
+                                                         size_t num) {
+    reserve(size() + num);
+    for (size_t row = 0; row < num; ++row) {
+        insert_data(data + offsets[row], offsets[row + 1] - offsets[row]);
+    }
+}
+
+void ColumnVarbinary::insert_many_dict_data(const int32_t* data_array, size_t start_index,
+                                            const StringRef* dict, size_t data_num,
+                                            uint32_t dict_num) {
+    reserve(size() + data_num);
+    // Decoder pages can be released after the call; copy long dictionary entries into our arena.
+    for (size_t row = start_index; row < start_index + data_num; ++row) {
+        const auto& value = dict[data_array[row]];
+        insert_data(value.data, value.size);
+    }
+}
+
 MutableColumnPtr ColumnVarbinary::clone_resized(size_t size) const {
     auto res = create();
     if (size > 0) {
diff --git a/be/src/core/column/column_varbinary.h b/be/src/core/column/column_varbinary.h
index caad77e28ad44f..2efc8da06fd048 100644
--- a/be/src/core/column/column_varbinary.h
+++ b/be/src/core/column/column_varbinary.h
@@ -30,6 +30,7 @@
 #include "core/string_view.h"
 
 namespace doris {
+// Binary IO does not enable hash computation; inherit IColumn's unsupported methods.
 class ColumnVarbinary final : public COWHelper<IColumn, ColumnVarbinary> {
 private:
     using Self = ColumnVarbinary;
@@ -189,6 +190,11 @@ class ColumnVarbinary final : public COWHelper<IColumn, ColumnVarbinary> {
     void insert_many_strings_overflow(const StringRef* strings, size_t num,
                                       size_t max_length) override;
 
+    void insert_many_continuous_binary_data(const char* data, const uint32_t* offsets,
+                                            size_t num) override;
+    void insert_many_dict_data(const int32_t* data_array, size_t start_index, const StringRef* dict,
+                               size_t data_num, uint32_t dict_num = 0) override;
+
     void sort_column(const ColumnSorter* sorter, EqualFlags& flags, IColumn::Permutation& perms,
                      EqualRange& range, bool last_column) const override;
 
diff --git a/be/src/core/data_type/data_type_factory.cpp b/be/src/core/data_type/data_type_factory.cpp
index 4b00064eb99318..8167c7a7a03996 100644
--- a/be/src/core/data_type/data_type_factory.cpp
+++ b/be/src/core/data_type/data_type_factory.cpp
@@ -635,6 +635,10 @@ DataTypePtr DataTypeFactory::create_data_type(
         } else if (primitive_type == TYPE_AGG_STATE) {
             // Do nothing
             nested = std::make_shared<DataTypeAggState>();
+        } else if (primitive_type == TYPE_VARBINARY) {
+            // Serialized execution types must retain VARBINARY(n)'s byte limit across RPCs.
+            return create_data_type(primitive_type, is_nullable, 0, 0,
+                                    scalar_type.has_len() ? scalar_type.len() : -1);
         } else if (primitive_type == TYPE_VARIANT) {
             nested = std::make_shared<DataTypeVariantV2>(node.variant_max_subcolumns_count(),
                                                          node.variant_enable_doc_mode());
diff --git a/be/src/core/data_type_serde/data_type_varbinary_serde.cpp b/be/src/core/data_type_serde/data_type_varbinary_serde.cpp
index 16ff1b20e8ac20..5832ffb1f307a4 100644
--- a/be/src/core/data_type_serde/data_type_varbinary_serde.cpp
+++ b/be/src/core/data_type_serde/data_type_varbinary_serde.cpp
@@ -18,13 +18,17 @@
 #include "core/data_type_serde/data_type_varbinary_serde.h"
 
 #include <cstring>
+#include <limits>
 
 #include "common/config.h"
 #include "core/column/column_varbinary.h"
 #include "core/data_type_serde/arrow_validation.h"
 #include "core/data_type_serde/parquet_decode_source.h"
+#include "exprs/function/string_hex_util.h"
+#include "util/url_coding.h"
 
 namespace doris {
+
 namespace {
 
 class VarbinaryParquetConsumer final : public ParquetFixedValueConsumer,
@@ -301,6 +305,82 @@ Status DataTypeVarbinarySerDe::deserialize_one_cell_from_json(IColumn& column, S
     return Status::OK();
 }
 
+Status DataTypeVarbinarySerDe::from_string(StringRef& str, IColumn& column,
+                                           const FormatOptions& options) const {
+    // Partition structs use the same hex representation as nested VARBINARY output. Decode it
+    // before appending so arbitrary bytes survive JSON transport instead of becoming NULL.
+    if (str.size < 2 || str.data[0] != '0' || str.data[1] != 'x' || (str.size - 2) % 2 != 0 ||
+        str.size - 2 > std::numeric_limits<int>::max()) {
+        return Status::InvalidArgument("Invalid VARBINARY hex representation");
+    }
+    // The INT_MAX guard also makes narrowing to the decoder's 32-bit offset type safe.
+    const auto hex_size = cast_set<ColumnString::Offset>(str.size - 2);
+    std::string bytes(hex_size / 2, '\0');
+    if (string_hex::hex_decode(str.data + 2, hex_size, bytes.data()) != bytes.size()) {
+        return Status::InvalidArgument("Invalid VARBINARY hex representation");
+    }
+    assert_cast<ColumnVarbinary&>(column).insert_data(bytes.data(), bytes.size());
+    return Status::OK();
+}
+
+Status DataTypeVarbinarySerDe::deserialize_one_cell_from_hive_text(
+        IColumn& column, Slice& slice, const FormatOptions& options,
+        int hive_text_complex_type_delimiter_level) const {
+    // Hive LazyBinary uses lenient Base64 (including URL-safe letters and whitespace),
+    // falling back to the original bytes for non-Base64 input or an empty decoding.
+    // Keep this separate from JSON/CSV: those formats do not share Hive's encoding contract.
+    std::string encoded;
+    encoded.reserve(slice.size);
+    bool padding = false;
+    for (size_t i = 0; i < slice.size; ++i) {
+        const char c = slice.data[i];
+        if (c == ' ' || c == '\t' || c == '\r' || c == '\n') {
+            continue;
+        }
+        if (c == '=') {
+            padding = true;
+            continue;
+        }
+        if (!((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') ||
+              c == '+' || c == '/' || c == '-' || c == '_')) {
+            return deserialize_one_cell_from_json(column, slice, options);
+        }
+        if (!padding) {
+            encoded.push_back(c == '-' ? '+' : c == '_' ? '/' : c);
+        }
+    }
+    // Commons Codec ignores a trailing sextet and accepts omitted padding.
+    if (encoded.size() % 4 == 1) {
+        encoded.pop_back();
+    }
+    encoded.append((4 - encoded.size() % 4) % 4, '=');
+    std::string decoded;
+    if (!base64_decode(encoded, &decoded) || decoded.empty()) {
+        return deserialize_one_cell_from_json(column, slice, options);
+    }
+    assert_cast<ColumnVarbinary&>(column).insert_data(decoded.data(), decoded.size());
+    return Status::OK();
+}
+
+Status DataTypeVarbinarySerDe::deserialize_column_from_hive_text_vector(
+        IColumn& column, std::vector<Slice>& slices, uint64_t* num_deserialized,
+        const FormatOptions& options, int hive_text_complex_type_delimiter_level) const {
+    DESERIALIZE_COLUMN_FROM_HIVE_TEXT_VECTOR()
+    return Status::OK();
+}
+
+Status DataTypeVarbinarySerDe::serialize_one_cell_to_hive_text(
+        const IColumn& column, int64_t row_num, BufferWritable& bw, FormatOptions& options,
+        int hive_text_complex_type_delimiter_level) const {
+    auto [data_column, data_row] = check_column_const_set_readability(column, row_num);
+    const auto value = assert_cast<const ColumnVarbinary&>(*data_column).get_data_at(data_row);
+    // Encoding is required on write as well, or a Hive reader will reinterpret binary bytes.
+    std::string encoded;
+    base64_encode(value.to_string(), &encoded);
+    bw.write(encoded.data(), encoded.size());
+    return Status::OK();
+}
+
 void DataTypeVarbinarySerDe::to_string(const IColumn& column, size_t row_num, BufferWritable& bw,
                                        const FormatOptions& options) const {
     const auto& value = assert_cast<const ColumnVarbinary&>(column).get_data()[row_num];
diff --git a/be/src/core/data_type_serde/data_type_varbinary_serde.h b/be/src/core/data_type_serde/data_type_varbinary_serde.h
index b4f5f26eab5f2d..4ac78c371ad4ff 100644
--- a/be/src/core/data_type_serde/data_type_varbinary_serde.h
+++ b/be/src/core/data_type_serde/data_type_varbinary_serde.h
@@ -49,6 +49,19 @@ class DataTypeVarbinarySerDe : public DataTypeSerDe {
     Status deserialize_one_cell_from_json(IColumn& column, Slice& slice,
                                           const FormatOptions& options) const override;
 
+    Status deserialize_one_cell_from_hive_text(
+            IColumn& column, Slice& slice, const FormatOptions& options,
+            int hive_text_complex_type_delimiter_level = 1) const override;
+
+    Status deserialize_column_from_hive_text_vector(
+            IColumn& column, std::vector<Slice>& slices, uint64_t* num_deserialized,
+            const FormatOptions& options,
+            int hive_text_complex_type_delimiter_level = 1) const override;
+
+    Status serialize_one_cell_to_hive_text(
+            const IColumn& column, int64_t row_num, BufferWritable& bw, FormatOptions& options,
+            int hive_text_complex_type_delimiter_level = 1) const override;
+
     Status deserialize_column_from_json_vector(IColumn& column, std::vector<Slice>& slices,
                                                uint64_t* num_deserialized,
                                                const FormatOptions& options) const override {
@@ -95,6 +108,9 @@ class DataTypeVarbinarySerDe : public DataTypeSerDe {
 
     void to_string(const IColumn& column, size_t row_num, BufferWritable& bw,
                    const FormatOptions& options) const override;
+
+    Status from_string(StringRef& str, IColumn& column,
+                       const FormatOptions& options) const override;
 };
 
 } // namespace doris
diff --git a/be/src/core/field.cpp b/be/src/core/field.cpp
index d6ee59009a6563..ec42f6b86c6afd 100644
--- a/be/src/core/field.cpp
+++ b/be/src/core/field.cpp
@@ -22,6 +22,8 @@
 
 #include "common/compare.h"
 #include "core/accurate_comparison.h"
+#include "core/allocator.h"
+#include "core/allocator_fwd.h"
 #include "core/data_type/data_type_decimal.h"
 #include "core/data_type/define_primitive_type.h"
 #include "core/data_type/primitive_type.h"
@@ -91,6 +93,50 @@ bool decimal_less_or_equal(Decimal128V3 x, Decimal128V3 y, UInt32 xs, UInt32 ys)
     return dec_less_or_equal<TYPE_DECIMAL128I>(x, y, xs, ys);
 }
 
+namespace {
+// Expression literals can outlive decoder pages and source columns.
+// Keep the view first for Field::get(), and fit ownership into the existing Field storage.
+struct OwnedBinaryField {
+    StringView view;
+    char* bytes = nullptr;
+    size_t byte_size = 0;
+
+    explicit OwnedBinaryField(const StringView& value) {
+        // Inline views already own their bytes; preserve their allocation-free representation.
+        if (value.isInline()) {
+            view = value;
+            return;
+        }
+        // Charge retained payloads and deep-copy peaks through Doris's checked allocator.
+        // Keep a standard-layout owner so the leading view remains accessible via Field::get().
+        bytes = static_cast<char*>(Allocator<false> {}.alloc(value.size()));
+        byte_size = value.size();
+        memcpy(bytes, value.data(), value.size());
+        view = StringView(bytes, value.size());
+    }
+    OwnedBinaryField(const OwnedBinaryField&) = delete;
+    OwnedBinaryField& operator=(const OwnedBinaryField&) = delete;
+    OwnedBinaryField& operator=(OwnedBinaryField&& other) noexcept {
+        release_bytes();
+        view = other.view;
+        bytes = std::exchange(other.bytes, nullptr);
+        byte_size = std::exchange(other.byte_size, 0);
+        return *this;
+    }
+    ~OwnedBinaryField() { release_bytes(); }
+
+private:
+    void release_bytes() const {
+        if (bytes != nullptr) {
+            // Field::get() exposes a mutable view; release the original allocation size.
+            Allocator<false> {}.free(bytes, byte_size);
+        }
+    }
+};
+static_assert(std::is_standard_layout_v<OwnedBinaryField>);
+static_assert(offsetof(OwnedBinaryField, view) == 0);
+} // namespace
+
 template <PrimitiveType Type>
 void Field::create_concrete(typename PrimitiveTypeTraits<Type>::CppType&& x) {
     // In both Field and PODArray, small types may be stored as wider types,
@@ -99,7 +145,12 @@ void Field::create_concrete(typename PrimitiveTypeTraits<Type>::CppType&& x) {
     // we must initialize the entire wide stored type, and not just the
     // nominal type.
     using StorageType = typename PrimitiveTypeTraits<Type>::CppType;
-    new (&storage) StorageType(std::move(x));
+    if constexpr (Type == TYPE_VARBINARY) {
+        static_assert(sizeof(OwnedBinaryField) <= sizeof(storage));
+        new (&storage) OwnedBinaryField(x);
+    } else {
+        new (&storage) StorageType(std::move(x));
+    }
     type = Type;
     DCHECK_NE(type, PrimitiveType::INVALID_TYPE);
 }
@@ -112,7 +163,11 @@ void Field::create_concrete(const typename PrimitiveTypeTraits<Type>::CppType& x
     // we must initialize the entire wide stored type, and not just the
     // nominal type.
     using StorageType = typename PrimitiveTypeTraits<Type>::CppType;
-    new (&storage) StorageType(x);
+    if constexpr (Type == TYPE_VARBINARY) {
+        new (&storage) OwnedBinaryField(x);
+    } else {
+        new (&storage) StorageType(x);
+    }
     type = Type;
     DCHECK_NE(type, PrimitiveType::INVALID_TYPE);
 }
@@ -241,6 +296,8 @@ Field& Field::operator=(const Field& rhs) {
     if (this != &rhs) {
         if (type != rhs.type) {
             destroy();
+            // A failed allocation while changing types must leave a destructible Field.
+            type = TYPE_NULL;
             create(rhs);
         } else {
             assign(rhs); /// This assigns string or vector without deallocation of existing buffer.
@@ -646,12 +703,20 @@ void Field::assign(const Field& field) {
 /// Assuming same types.
 template <PrimitiveType Type>
 void Field::assign_concrete(typename PrimitiveTypeTraits<Type>::CppType&& x) {
+    if constexpr (Type == TYPE_VARBINARY) {
+        *reinterpret_cast<OwnedBinaryField*>(&storage) = OwnedBinaryField(x);
+        return;
+    }
     auto* MAY_ALIAS ptr = reinterpret_cast<typename PrimitiveTypeTraits<Type>::CppType*>(&storage);
     *ptr = std::forward<typename PrimitiveTypeTraits<Type>::CppType>(x);
 }
 
 template <PrimitiveType Type>
 void Field::assign_concrete(const typename PrimitiveTypeTraits<Type>::CppType& x) {
+    if constexpr (Type == TYPE_VARBINARY) {
+        *reinterpret_cast<OwnedBinaryField*>(&storage) = OwnedBinaryField(x);
+        return;
+    }
     auto* MAY_ALIAS ptr = reinterpret_cast<typename PrimitiveTypeTraits<Type>::CppType*>(&storage);
     *ptr = std::forward<const typename PrimitiveTypeTraits<Type>::CppType>(x);
 }
@@ -683,6 +748,10 @@ const typename PrimitiveTypeTraits<T>::CppType& Field::get() const {
 
 template <PrimitiveType T>
 void Field::destroy() {
+    if constexpr (T == TYPE_VARBINARY) {
+        reinterpret_cast<OwnedBinaryField*>(&storage)->~OwnedBinaryField();
+        return;
+    }
     using TargetType = typename PrimitiveTypeTraits<T>::CppType;
     DCHECK(T == type || ((is_string_type(type) && is_string_type(T))))
             << "Type mismatch: requested " << type_to_string(T) << ", actual " << get_type_name();
diff --git a/be/src/core/field.h b/be/src/core/field.h
index cf350b4c4690ea..155278fa337919 100644
--- a/be/src/core/field.h
+++ b/be/src/core/field.h
@@ -191,13 +191,15 @@ class Field {
     Field(PrimitiveType w) : type(w) {}
     template <PrimitiveType T>
     static Field create_field(const typename PrimitiveTypeTraits<T>::CppType& data) {
-        auto f = Field(T);
+        // Publish the type only after construction succeeds, so allocation failures cannot
+        // destroy uninitialized owned storage (including long binary values).
+        auto f = Field();
         f.template create_concrete<T>(data);
         return f;
     }
     template <PrimitiveType T>
     static Field create_field(typename PrimitiveTypeTraits<T>::CppType&& data) {
-        auto f = Field(T);
+        auto f = Field();
         f.template create_concrete<T>(std::move(data));
         return f;
     }
@@ -243,6 +245,7 @@ class Field {
         if (this != &rhs) {
             if (type != rhs.type) {
                 destroy();
+                type = TYPE_NULL;
                 create(std::move(rhs));
             } else {
                 assign(std::move(rhs));
diff --git a/be/src/core/value/timestamptz_value.cpp b/be/src/core/value/timestamptz_value.cpp
index ffa9bf530e2809..05114342b30568 100644
--- a/be/src/core/value/timestamptz_value.cpp
+++ b/be/src/core/value/timestamptz_value.cpp
@@ -17,6 +17,7 @@
 
 #include "core/value/timestamptz_value.h"
 
+#include "common/exception.h"
 #include "exprs/function/cast/cast_to_timestamptz_impl.hpp"
 
 namespace doris {
@@ -38,6 +39,13 @@ std::string TimestampTzValue::to_string(const cctz::time_zone& tz, int scale) co
     auto lookup_result = tz.lookup(cur_tz_time);
 
     cctz::civil_second civ = lookup_result.cs;
+    // UTC storage bounds do not guarantee a representable session-local year. Reject
+    // overflow before DateTimeV2 formatting could produce an offset-only wire value.
+    if (civ.year() < 0 || civ.year() > 9999) {
+        throw Exception(ErrorCode::INVALID_ARGUMENT,
+                        "TIMESTAMPTZ local year is outside [0, 9999]: year={}, timezone={}",
+                        civ.year(), tz.name());
+    }
     auto time_offset = lookup_result.offset;
 
     bool is_negative_offset = time_offset < 0;
@@ -65,6 +73,14 @@ std::string TimestampTzValue::to_string(const cctz::time_zone& tz, int scale) co
     buffer[len++] = ':';
     buffer[len++] = static_cast<char>('0' + offset_mins / 10);
     buffer[len++] = '0' + offset_mins % 10;
+    // Historical zones can have sub-minute offsets. Dropping their seconds changes the
+    // instant represented by the client-visible wall clock and offset when read back.
+    const int offset_seconds = abs_offset % 60;
+    if (offset_seconds != 0) {
+        buffer[len++] = ':';
+        buffer[len++] = static_cast<char>('0' + offset_seconds / 10);
+        buffer[len++] = static_cast<char>('0' + offset_seconds % 10);
+    }
     return {buffer, static_cast<size_t>(len)};
 }
 
diff --git a/be/src/exec/common/hash_table/hash_key_type.h b/be/src/exec/common/hash_table/hash_key_type.h
index 8ce7882f3a6bde..09bd3d606427be 100644
--- a/be/src/exec/common/hash_table/hash_key_type.h
+++ b/be/src/exec/common/hash_table/hash_key_type.h
@@ -102,6 +102,13 @@ inline HashKeyType get_hash_key_type_fixed(const std::vector<DataTypePtr>& data_
 }
 
 inline HashKeyType get_hash_key_type(const std::vector<DataTypePtr>& data_types) {
+    // Reject binary before the multi-key serialization fallback can enable joins or grouping.
+    for (const auto& type : data_types) {
+        if (type->get_primitive_type() == TYPE_VARBINARY) {
+            throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+                            "VARBINARY hash keys are not supported");
+        }
+    }
     if (data_types.size() > 1) {
         return get_hash_key_type_fixed(data_types);
     }
diff --git a/be/src/exec/sink/writer/iceberg/partition_transformers.cpp b/be/src/exec/sink/writer/iceberg/partition_transformers.cpp
index e38e0e4c5eb8a7..d52ea289df1a2c 100644
--- a/be/src/exec/sink/writer/iceberg/partition_transformers.cpp
+++ b/be/src/exec/sink/writer/iceberg/partition_transformers.cpp
@@ -46,6 +46,12 @@ const std::chrono::sys_days PartitionColumnTransformUtils::EPOCH = std::chrono::
 std::unique_ptr<PartitionColumnTransform> PartitionColumnTransforms::create(
         const doris::iceberg::PartitionField& field, const DataTypePtr& source_type) {
     auto& transform = field.transform();
+    // Identity/void only carry values; computed binary partition transforms are unsupported.
+    if (source_type->get_primitive_type() == TYPE_VARBINARY && transform != "identity" &&
+        transform != "void") {
+        throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+                        "VARBINARY partition transform {} is not supported", transform);
+    }
     static const std::regex has_width(R"((\w+)\[(\d+)\])");
     std::smatch width_match;
 
diff --git a/be/src/exprs/aggregate/aggregate_function_min_max_impl.h b/be/src/exprs/aggregate/aggregate_function_min_max_impl.h
index 9717cc0461c7c1..0c49c79aa8a2e9 100644
--- a/be/src/exprs/aggregate/aggregate_function_min_max_impl.h
+++ b/be/src/exprs/aggregate/aggregate_function_min_max_impl.h
@@ -141,6 +141,10 @@ AggregateFunctionPtr create_aggregate_function_single_value(const String& name,
         return creator_without_type::create_unary_arguments<
                 AggregateFunctionsSingleValue<Data<SingleValueDataComplexType>>>(
                 argument_types, result_is_nullable, attr);
+    case PrimitiveType::TYPE_VARBINARY:
+        // Owning binary values for IO must not implicitly enable single-value aggregates.
+        throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "VARBINARY aggregate {} is not supported",
+                        name);
     default:
         return nullptr;
     }
diff --git a/be/src/exprs/create_predicate_function.h b/be/src/exprs/create_predicate_function.h
index 43c32ebb0b59fd..afdfe80e79837e 100644
--- a/be/src/exprs/create_predicate_function.h
+++ b/be/src/exprs/create_predicate_function.h
@@ -108,6 +108,9 @@ typename Traits::BasePtr create_predicate_function(PrimitiveType type, bool null
     using Creator = PredicateFunctionCreator<Traits>;
 
     switch (type) {
+    case TYPE_VARBINARY:
+        // Binary read/write support does not provide storage or runtime predicate kernels.
+        throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR, "VARBINARY predicates are not supported");
     case TYPE_BOOLEAN: {
         return Creator::template create<TYPE_BOOLEAN, N>(null_aware);
     }
diff --git a/be/src/exprs/function/cast/cast_to_date.h b/be/src/exprs/function/cast/cast_to_date.h
index eecfcb6d552fb2..dad009409a6ae6 100644
--- a/be/src/exprs/function/cast/cast_to_date.h
+++ b/be/src/exprs/function/cast/cast_to_date.h
@@ -504,9 +504,11 @@ class CastToImpl<CastModeType::StrictMode, DataTypeTimeStampTz, DataTypeDateTime
             TimestampTzValue from_tz {col_from[i]};
             DateV2Value<DateTimeV2ValueType> dt;
             if (!from_tz.to_datetime(dt, local_time_zone, dt_scale, tz_scale)) {
+                // The failed local conversion may also be unformattable. Render the stored
+                // UTC fields so reporting the cast error cannot throw a second exception.
                 return Status::InvalidArgument(
-                        "can not cast from  timestamptz : {} to datetime in timezone : {}",
-                        from_tz.to_string(local_time_zone), context->state()->timezone());
+                        "can not cast from  timestamptz : {} UTC to datetime in timezone : {}",
+                        from_tz.utc_dt().to_string(tz_scale), context->state()->timezone());
             }
             col_to_data[i] = dt.to_date_int_val();
         }
diff --git a/be/src/exprs/function/cast/cast_to_datetimev2_impl.hpp b/be/src/exprs/function/cast/cast_to_datetimev2_impl.hpp
index 21c78ca0f09167..9c1999d2eadd9d 100644
--- a/be/src/exprs/function/cast/cast_to_datetimev2_impl.hpp
+++ b/be/src/exprs/function/cast/cast_to_datetimev2_impl.hpp
@@ -695,6 +695,7 @@ inline bool CastToDatetimeV2::from_string_strict_mode_internal(
             const char sign = *ptr;
             ++ptr;
             part[1] = 0;
+            uint32_t second_offset = 0;
 
             uint32_t length = count_digits(ptr, end);
             // hour
@@ -705,7 +706,9 @@ inline bool CastToDatetimeV2::from_string_strict_mode_internal(
                 SET_PARAMS_RET_FALSE_IFN((consume_digit<UInt32, 2>(ptr, end, part[0])),
                                          "invalid hour offset '{}'", std::string {ptr, end});
             }
-            SET_PARAMS_RET_FALSE_IFN(part[0] <= 14, "invalid hour offset '{}'", part[0]);
+            SET_PARAMS_RET_FALSE_IFN(
+                    part[0] < (type == DataTimeCastEnumType::TIMESTAMP_TZ ? 24U : 15U),
+                    "invalid hour offset '{}'", part[0]);
             if (ptr < end) {
                 if (*ptr == ':') {
                     ++ptr;
@@ -713,16 +716,37 @@ inline bool CastToDatetimeV2::from_string_strict_mode_internal(
                 // minute
                 SET_PARAMS_RET_FALSE_IFN((consume_digit<UInt32, 2>(ptr, end, part[1])),
                                          "invalid minute offset '{}'", std::string {ptr, end});
-                SET_PARAMS_RET_FALSE_IFN((part[1] == 0 || part[1] == 30 || part[1] == 45),
-                                         "invalid minute offset '{}'", part[1]);
+                if constexpr (type == DataTimeCastEnumType::TIMESTAMP_TZ) {
+                    // TIMESTAMPTZ output preserves historical offsets, including seconds and
+                    // non-quarter-hour minutes. Keep the legacy DATETIME parser unchanged.
+                    SET_PARAMS_RET_FALSE_IFN(part[1] < 60, "invalid minute offset '{}'", part[1]);
+                    if (ptr < end && *ptr == ':') {
+                        ++ptr;
+                        SET_PARAMS_RET_FALSE_IFN(
+                                (consume_digit<UInt32, 2>(ptr, end, second_offset)),
+                                "invalid second offset '{}'", std::string {ptr, end});
+                        SET_PARAMS_RET_FALSE_IFN(second_offset < 60, "invalid second offset '{}'",
+                                                 second_offset);
+                    }
+                } else {
+                    SET_PARAMS_RET_FALSE_IFN((part[1] == 0 || part[1] == 30 || part[1] == 45),
+                                             "invalid minute offset '{}'", part[1]);
+                }
+            }
+            if constexpr (type == DataTimeCastEnumType::TIMESTAMP_TZ) {
+                // Wire offsets include historical zones outside the session fixed-zone range.
+                // Use the exact offset even when it has no seconds (for example, Guam's -14:21).
+                const auto offset = static_cast<int>(part[0] * 3600 + part[1] * 60 + second_offset);
+                parsed_tz = cctz::fixed_time_zone(cctz::seconds(sign == '-' ? -offset : offset));
+            } else {
+                SET_PARAMS_RET_FALSE_IFN(part[0] != 14 || part[1] == 0,
+                                         "invalid timezone offset '{}'",
+                                         combine_tz_offset(sign, part[0], part[1]));
+                SET_PARAMS_RET_FALSE_IFN(
+                        TimezoneUtils::find_cctz_time_zone(
+                                combine_tz_offset(sign, part[0], part[1]), parsed_tz),
+                        "invalid timezone offset '{}'", combine_tz_offset(sign, part[0], part[1]));
             }
-            SET_PARAMS_RET_FALSE_IFN(part[0] != 14 || part[1] == 0, "invalid timezone offset '{}'",
-                                     combine_tz_offset(sign, part[0], part[1]));
-
-            SET_PARAMS_RET_FALSE_IFN(TimezoneUtils::find_cctz_time_zone(
-                                             combine_tz_offset(sign, part[0], part[1]), parsed_tz),
-                                     "invalid timezone offset '{}'",
-                                     combine_tz_offset(sign, part[0], part[1]));
         } else {
             // timezone name
             const auto* start = ptr;
@@ -959,7 +983,7 @@ inline bool CastToDatetimeV2::from_string_non_strict_mode_internal(
             // offset
             const char sign = *ptr;
             ++ptr;
-            uint32_t hour_offset, minute_offset = 0;
+            uint32_t hour_offset, minute_offset = 0, second_offset = 0;
 
             uint32_t length = count_digits(ptr, end);
             // hour
@@ -968,26 +992,45 @@ inline bool CastToDatetimeV2::from_string_non_strict_mode_internal(
             } else {
                 PROPAGATE_FALSE((consume_digit<UInt32, 2>(ptr, end, hour_offset)));
             }
-            SET_PARAMS_RET_FALSE_IFN(hour_offset <= 14, "invalid hour offset '{}'", hour_offset);
+            SET_PARAMS_RET_FALSE_IFN(
+                    hour_offset < (type == DataTimeCastEnumType::TIMESTAMP_TZ ? 24U : 15U),
+                    "invalid hour offset '{}'", hour_offset);
             if (ptr < end) {
                 if (*ptr == ':') {
                     ++ptr;
                 }
                 // minute
                 PROPAGATE_FALSE((consume_digit<UInt32, 2>(ptr, end, minute_offset)));
+                if constexpr (type == DataTimeCastEnumType::TIMESTAMP_TZ) {
+                    SET_PARAMS_RET_FALSE_IFN(minute_offset < 60, "invalid minute offset '{}'",
+                                             minute_offset);
+                    if (ptr < end && *ptr == ':') {
+                        ++ptr;
+                        PROPAGATE_FALSE((consume_digit<UInt32, 2>(ptr, end, second_offset)));
+                        SET_PARAMS_RET_FALSE_IFN(second_offset < 60, "invalid second offset '{}'",
+                                                 second_offset);
+                    }
+                } else {
+                    SET_PARAMS_RET_FALSE_IFN(
+                            (minute_offset == 0 || minute_offset == 30 || minute_offset == 45),
+                            "invalid minute offset {}", minute_offset);
+                }
+            }
+            if constexpr (type == DataTimeCastEnumType::TIMESTAMP_TZ) {
+                // Match strict parsing: a serialized historical offset is not a session setting.
+                const auto offset =
+                        static_cast<int>(hour_offset * 3600 + minute_offset * 60 + second_offset);
+                parsed_tz = cctz::fixed_time_zone(cctz::seconds(sign == '-' ? -offset : offset));
+            } else {
+                SET_PARAMS_RET_FALSE_IFN(hour_offset != 14 || minute_offset == 0,
+                                         "invalid timezone offset '{}'",
+                                         combine_tz_offset(sign, hour_offset, minute_offset));
                 SET_PARAMS_RET_FALSE_IFN(
-                        (minute_offset == 0 || minute_offset == 30 || minute_offset == 45),
-                        "invalid minute offset {}", minute_offset);
+                        TimezoneUtils::find_cctz_time_zone(
+                                combine_tz_offset(sign, hour_offset, minute_offset), parsed_tz),
+                        "invalid timezone offset '{}'",
+                        combine_tz_offset(sign, hour_offset, minute_offset));
             }
-            SET_PARAMS_RET_FALSE_IFN(hour_offset != 14 || minute_offset == 0,
-                                     "invalid timezone offset '{}'",
-                                     combine_tz_offset(sign, hour_offset, minute_offset));
-
-            SET_PARAMS_RET_FALSE_IFN(
-                    TimezoneUtils::find_cctz_time_zone(
-                            combine_tz_offset(sign, hour_offset, minute_offset), parsed_tz),
-                    "invalid timezone offset '{}'",
-                    combine_tz_offset(sign, hour_offset, minute_offset));
         } else {
             // timezone name
             const auto* start = ptr;
diff --git a/be/src/exprs/function/cast/cast_to_string.h b/be/src/exprs/function/cast/cast_to_string.h
index 3e4e188b7ccf90..8f205386c76c8d 100644
--- a/be/src/exprs/function/cast/cast_to_string.h
+++ b/be/src/exprs/function/cast/cast_to_string.h
@@ -17,6 +17,8 @@
 
 #pragma once
 
+#include <algorithm>
+
 #include "core/data_type_serde/data_type_serde.h"
 #include "core/types.h"
 #include "core/value/time_value.h"
@@ -581,7 +583,22 @@ class CastToStringFunction {
             limited_col = col_from.cut(0, input_rows_count);
             col_to_serialize = limited_col.get();
         }
-        type.get_serde()->to_string_batch(*col_to_serialize, *col_to, options);
+        const auto serde = type.get_serde();
+        if (null_map != nullptr && std::any_of(null_map, null_map + input_rows_count,
+                                               [](auto value) { return value != 0; })) {
+            // Nested payloads of NULL rows may be uninitialized or outside the type's
+            // domain. Do not format them before the nullable wrapper restores the mask.
+            col_to->reserve(input_rows_count);
+            VectorBufferWriter write_buffer(*col_to);
+            for (size_t row = 0; row < input_rows_count; ++row) {
+                if (!null_map[row]) {
+                    serde->to_string(*col_to_serialize, row, write_buffer, options);
+                }
+                write_buffer.commit();
+            }
+        } else {
+            serde->to_string_batch(*col_to_serialize, *col_to, options);
+        }
 
         block.replace_by_position(result, std::move(col_to));
         return Status::OK();
diff --git a/be/src/exprs/function/cast/cast_to_timestamp_ns.h b/be/src/exprs/function/cast/cast_to_timestamp_ns.h
index 2ba379d940b979..7e6645539674bf 100644
--- a/be/src/exprs/function/cast/cast_to_timestamp_ns.h
+++ b/be/src/exprs/function/cast/cast_to_timestamp_ns.h
@@ -382,9 +382,10 @@ class CastToImpl<CastMode, DataTypeTimeStampTz, DataTypeTimeStampNs> : public Ca
                     col_to->get_data()[i].from_datetime(datetime);
             if (!converted) {
                 if constexpr (CastMode == CastModeType::StrictMode) {
+                    // The session-local year may be unrepresentable even though UTC is valid.
                     return Status::InvalidArgument(
                             "can not cast timestamptz {} to TIMESTAMP_NS in timezone {}",
-                            source.to_string(local_time_zone), context->state()->timezone());
+                            source.utc_dt().to_string(source_scale), context->state()->timezone());
                 }
                 col_null->get_data()[i] = true;
             }
diff --git a/be/src/exprs/function/cast/cast_to_timestamptz.h b/be/src/exprs/function/cast/cast_to_timestamptz.h
index 1e31cf24be5975..f09bf602a034b1 100644
--- a/be/src/exprs/function/cast/cast_to_timestamptz.h
+++ b/be/src/exprs/function/cast/cast_to_timestamptz.h
@@ -200,7 +200,6 @@ class CastToImpl<CastModeType::StrictMode, DataTypeTimeStampTz, DataTypeTimeStam
 
         auto col_to = ColumnTimeStampTz::create(input_rows_count);
         auto& col_to_data = col_to->get_data();
-        const auto& local_time_zone = context->state()->timezone_obj();
 
         const auto from_scale = block.get_by_position(arguments[0]).type->get_scale();
         const auto to_scale = block.get_by_position(result).type->get_scale();
@@ -214,10 +213,11 @@ class CastToImpl<CastModeType::StrictMode, DataTypeTimeStampTz, DataTypeTimeStam
             auto& to_tz = col_to_data[i];
 
             if (!transform_date_scale(to_scale, from_scale, to_tz, from_tz)) {
+                // Error reporting must not format an overflowing timestamp in a session
+                // timezone whose local year may itself be outside the supported range.
                 return Status::InvalidArgument(
-                        "can not cast from  timestamptz : {} to timestamptz in timezone : {}",
-                        TimestampTzValue {from_tz}.to_string(local_time_zone, from_scale),
-                        context->state()->timezone());
+                        "can not cast from  timestamptz : {} UTC to timestamptz in timezone : {}",
+                        from_tz.utc_dt().to_string(from_scale), context->state()->timezone());
             }
         }
         block.get_by_position(result).column = std::move(col_to);
diff --git a/be/src/exprs/function/functions_comparison.h b/be/src/exprs/function/functions_comparison.h
index ffb6958be4d7f1..0bfce7e67f5072 100644
--- a/be/src/exprs/function/functions_comparison.h
+++ b/be/src/exprs/function/functions_comparison.h
@@ -703,10 +703,10 @@ class FunctionComparison : public IFunction {
                 const auto scale = temporal_type->get_scale();
                 if (!temporal.to_datetime(local_datetime, context->state()->timezone_obj(), scale,
                                           scale)) [[unlikely]] {
+                    // Preserve the comparison error instead of re-entering the failing formatter.
                     return Status::InvalidArgument(
                             "can not compare timestamptz {} with TIMESTAMP_NS in timezone {}",
-                            temporal.to_string(context->state()->timezone_obj(), scale),
-                            context->state()->timezone());
+                            temporal.utc_dt().to_string(scale), context->state()->timezone());
                 }
                 comparison = compare_timestamp_ns_with_temporal(timestamp, local_datetime);
             } else {
diff --git a/be/src/exprs/function/in.h b/be/src/exprs/function/in.h
index 9075415e6365cb..10e101324f6a79 100644
--- a/be/src/exprs/function/in.h
+++ b/be/src/exprs/function/in.h
@@ -105,6 +105,10 @@ class FunctionIn : public IFunction {
         if (scope == FunctionContext::THREAD_LOCAL) {
             return Status::OK();
         }
+        // Binary IO must not route IN through the shared string/storage predicate implementation.
+        if (context->get_arg_type(0)->get_primitive_type() == TYPE_VARBINARY) {
+            return Status::NotSupported("VARBINARY IN/NOT IN is not supported");
+        }
         std::shared_ptr<InState> state = std::make_shared<InState>();
         context->set_function_state(scope, state);
         DCHECK(context->get_num_args() >= 1);
diff --git a/be/src/exprs/vin_predicate.cpp b/be/src/exprs/vin_predicate.cpp
index 64a0f60a405901..d956c2173a0e6a 100644
--- a/be/src/exprs/vin_predicate.cpp
+++ b/be/src/exprs/vin_predicate.cpp
@@ -176,7 +176,9 @@ void VInPredicate::_prepare_zonemap_min_max(VExprContext* context) {
     // dictionary, and raw evaluation direct-slot-only while Bloom may consume a nested leaf.
     const auto data_type = remove_nullable(bloom_probe->value_type);
     DORIS_CHECK(data_type != nullptr);
-    if (is_complex_type(data_type->get_primitive_type())) {
+    // Binary IN is rejected by the SQL function; do not build storage predicates for its keys.
+    if (is_complex_type(data_type->get_primitive_type()) ||
+        data_type->get_primitive_type() == TYPE_VARBINARY) {
         return;
     }
 
diff --git a/be/src/util/raw_value.h b/be/src/util/raw_value.h
index ca9914e4064d9d..4ed839e0bc5e49 100644
--- a/be/src/util/raw_value.h
+++ b/be/src/util/raw_value.h
@@ -23,6 +23,7 @@
 #include <string>
 
 #include "common/consts.h"
+#include "common/exception.h"
 #include "common/logging.h"
 #include "core/data_type/define_primitive_type.h"
 #include "core/packed_int128.h"
@@ -44,6 +45,11 @@ class RawValue {
 // Because crc32 hardware is not equal with zlib crc32
 inline uint32_t RawValue::zlib_crc32(const void* v, size_t len, const PrimitiveType& type,
                                      uint32_t seed) {
+    // Reject binary even for NULL instead of reaching the default-type assertion or hash path.
+    if (type == TYPE_VARBINARY) {
+        throw Exception(ErrorCode::NOT_IMPLEMENTED_ERROR,
+                        "VARBINARY tablet routing hash is not supported");
+    }
     // Hash_combine with v = 0
     if (v == nullptr) {
         uint32_t value = 0x9e3779b9;
diff --git a/be/src/util/timezone_utils.h b/be/src/util/timezone_utils.h
index 3ae0a23d1d1402..1f37c7e2e9efdf 100644
--- a/be/src/util/timezone_utils.h
+++ b/be/src/util/timezone_utils.h
@@ -18,6 +18,7 @@
 
 #pragma once
 
+#include <cstdint>
 #include <string>
 
 namespace cctz {
diff --git a/be/test/core/column/column_varbinary_test.cpp b/be/test/core/column/column_varbinary_test.cpp
index 0360b14cda8ab7..20e07dfa8af27f 100644
--- a/be/test/core/column/column_varbinary_test.cpp
+++ b/be/test/core/column/column_varbinary_test.cpp
@@ -36,9 +36,138 @@
 #include "core/string_ref.h"
 #include "core/string_view.h"
 #include "core/types.h"
+#include "exec/common/sip_hash.h"
+#include "runtime/memory/mem_tracker_limiter.h"
+#include "runtime/thread_context.h"
+#include "util/defer_op.h"
+#include "util/raw_value.h"
 
 namespace doris {
 
+TEST(ColumnVarbinaryStorageTest, FieldsOwnValuesAcrossInlineBoundary) {
+    for (size_t size : {0U, 1U, 12U, 13U, 64U}) {
+        SCOPED_TRACE(size);
+        const std::string expected(size, '\xff');
+        std::string source = expected;
+        auto field = Field::create_field<TYPE_VARBINARY>(StringView(source));
+        Field copied = field;
+        source.assign(size, 'x');
+        EXPECT_EQ(field.get<TYPE_VARBINARY>().str(), expected);
+        EXPECT_EQ(copied.get<TYPE_VARBINARY>().str(), expected);
+        field = Field::create_field<TYPE_VARBINARY>(StringView("replacement"));
+        Field moved = std::move(copied);
+        EXPECT_EQ(moved.get<TYPE_VARBINARY>().str(), expected);
+        moved = field;
+        EXPECT_EQ(moved.get<TYPE_VARBINARY>().str(), "replacement");
+    }
+}
+
+TEST(ColumnVarbinaryStorageTest, FieldsOwnLongBinaryValues) {
+    const std::string expected(64, '\xff');
+    Field copy;
+    {
+        auto column = ColumnVarbinary::create();
+        column->insert_data(expected.data(), expected.size());
+        Field value = (*column)[0];
+        EXPECT_NE(value.get<TYPE_VARBINARY>().data(), column->get_data_at(0).data);
+        copy = value;
+        EXPECT_NE(copy.get<TYPE_VARBINARY>().data(), value.get<TYPE_VARBINARY>().data());
+        column->clear();
+    }
+    EXPECT_EQ(copy.get<TYPE_VARBINARY>().str(), expected);
+    copy = Field::create_field<TYPE_VARBINARY>(StringView("a"));
+    EXPECT_EQ(copy.get<TYPE_VARBINARY>().str(), "a");
+}
+
+TEST(ColumnVarbinaryStorageTest, FieldsChargeAndReleaseTrackedMemory) {
+    const std::string payload(64, '\xff');
+    const std::string smaller(16, 's');
+    auto tracker = MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+                                                    "binary-field-ownership", 1024);
+    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker);
+    const auto consumption = [&] {
+        thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+        return tracker->consumption();
+    };
+    {
+        auto field = Field::create_field<TYPE_VARBINARY>(StringView(payload));
+        EXPECT_EQ(consumption(), 64);
+        Field copied = field;
+        EXPECT_EQ(consumption(), 128);
+        // Replacement must release the old allocation using its original size.
+        copied = Field::create_field<TYPE_VARBINARY>(StringView(smaller));
+        EXPECT_EQ(consumption(), 80);
+        EXPECT_EQ(copied.get<TYPE_VARBINARY>().str(), smaller);
+        copied = Field::create_field<TYPE_VARBINARY>(StringView("inline"));
+        EXPECT_EQ(consumption(), 64);
+        auto inline_field = Field::create_field<TYPE_VARBINARY>(StringView("inline"));
+        EXPECT_EQ(consumption(), 64);
+        EXPECT_EQ(inline_field.get<TYPE_VARBINARY>().str(), "inline");
+        field.get<TYPE_VARBINARY>() = StringView("shorter view");
+        EXPECT_EQ(consumption(), 64);
+    }
+    EXPECT_EQ(consumption(), 0);
+}
+
+TEST(ColumnVarbinaryStorageTest, FieldsRespectTrackedMemoryLimit) {
+    const std::string payload(64, '\xff');
+    const std::string oversized(128, 'x');
+    auto tracker = MemTrackerLimiter::create_shared(MemTrackerLimiter::Type::OTHER,
+                                                    "binary-field-limit", 96);
+    SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(tracker);
+    ++enable_thread_catch_bad_alloc;
+    Defer restore_catch_bad_alloc {[] { --enable_thread_catch_bad_alloc; }};
+    const auto consumption = [&] {
+        thread_context()->thread_mem_tracker_mgr->flush_untracked_mem();
+        return tracker->consumption();
+    };
+    const auto expect_allocation_failure = [](auto&& operation) {
+        try {
+            operation();
+            FAIL() << "Expected the binary payload to respect the memory limit";
+        } catch (const Exception& e) {
+            EXPECT_EQ(e.code(), ErrorCode::MEM_ALLOC_FAILED);
+        }
+    };
+    expect_allocation_failure(
+            [&] { auto field = Field::create_field<TYPE_VARBINARY>(StringView(oversized)); });
+    EXPECT_EQ(consumption(), 0);
+    {
+        auto field = Field::create_field<TYPE_VARBINARY>(StringView(payload));
+        EXPECT_EQ(consumption(), 64);
+        // A deep copy must check the peak while the source is still retained.
+        expect_allocation_failure([&] { Field copied = field; });
+        EXPECT_EQ(consumption(), 64);
+        auto destination = Field::create_field<TYPE_VARBINARY>(StringView("inline"));
+        expect_allocation_failure([&] { destination = field; });
+        EXPECT_EQ(destination.get<TYPE_VARBINARY>().str(), "inline");
+        EXPECT_EQ(field.get<TYPE_VARBINARY>().str(), payload);
+        EXPECT_EQ(consumption(), 64);
+        const std::string smaller(16, 's');
+        destination = Field::create_field<TYPE_VARBINARY>(StringView(smaller));
+        EXPECT_EQ(consumption(), 80);
+        expect_allocation_failure([&] { destination = field; });
+        EXPECT_EQ(destination.get<TYPE_VARBINARY>().str(), smaller);
+        EXPECT_EQ(consumption(), 80);
+    }
+    EXPECT_EQ(consumption(), 0);
+}
+
+TEST(ColumnVarbinaryStorageTest, StorageDecoderInsertionPreservesBinaryPayloads) {
+    auto column = ColumnVarbinary::create();
+    const std::string payload("\0a\0\xff", 4);
+    const uint32_t offsets[] = {0, 0, 1, 4};
+    ASSERT_NO_THROW(column->insert_many_continuous_binary_data(payload.data(), offsets, 3));
+    EXPECT_EQ(column->get_data_at(0).to_string(), "");
+    EXPECT_EQ(column->get_data_at(1).to_string(), std::string("\0", 1));
+    EXPECT_EQ(column->get_data_at(2).to_string(), payload.substr(1));
+    const StringRef dictionary[] = {{payload.data(), payload.size()}, {"", 0}};
+    const int32_t codes[] = {1, 0, 1};
+    ASSERT_NO_THROW(column->insert_many_dict_data(codes, 1, dictionary, 2, 2));
+    EXPECT_EQ(column->get_data_at(3).to_string(), payload);
+    EXPECT_EQ(column->get_data_at(4).to_string(), "");
+}
+
 class ColumnVarbinaryTest : public ::testing::Test {
 protected:
     void SetUp() override {}
@@ -113,6 +242,31 @@ TEST_F(ColumnVarbinaryTest, BasicInsertGetPopClear) {
     EXPECT_EQ(col->byte_size(), 0U);
 }
 
+TEST_F(ColumnVarbinaryTest, TabletRoutingHashIsNotSupported) {
+    for (const char* value : {static_cast<const char*>(nullptr), "", "binary"}) {
+        EXPECT_THROW(RawValue::zlib_crc32(value, value == nullptr ? 0 : strlen(value),
+                                          TYPE_VARBINARY, 0),
+                     Exception);
+    }
+}
+
+TEST_F(ColumnVarbinaryTest, HashingIsNotSupported) {
+    auto binary = ColumnVarbinary::create();
+    binary->insert_data("\0\xff", 2);
+    SipHash sip;
+    uint64_t hash64 = 17;
+    uint32_t hash32 = 23;
+    EXPECT_THROW(binary->update_hash_with_value(0, sip), Exception);
+    EXPECT_THROW(binary->update_hashes_with_value(&hash64, nullptr), Exception);
+    EXPECT_THROW(binary->update_xxHash_with_value(0, 1, hash64, nullptr), Exception);
+    EXPECT_THROW(binary->update_crcs_with_value(&hash32, TYPE_VARBINARY, 1, 0, nullptr), Exception);
+    EXPECT_THROW(binary->update_crc_with_value(0, 1, hash32, nullptr), Exception);
+    EXPECT_THROW(binary->update_crc32c_batch(&hash32, nullptr), Exception);
+    EXPECT_THROW(binary->update_crc32c_single(0, 1, hash32, nullptr), Exception);
+    EXPECT_EQ(hash64, 17);
+    EXPECT_EQ(hash32, 23);
+}
+
 TEST_F(ColumnVarbinaryTest, InsertFromAndRanges) {
     auto src = ColumnVarbinary::create();
     std::vector<std::string> vals = {make_bytes(1, 0x01), make_bytes(2, 0x02),
diff --git a/be/test/core/data_type/data_type_varbinary_test.cpp b/be/test/core/data_type/data_type_varbinary_test.cpp
index d71710ceb25ce0..3e4ee03ced2f05 100644
--- a/be/test/core/data_type/data_type_varbinary_test.cpp
+++ b/be/test/core/data_type/data_type_varbinary_test.cpp
@@ -35,12 +35,14 @@
 #include "core/column/column_varbinary.h"
 #include "core/data_type/common_data_type_serder_test.h"
 #include "core/data_type/common_data_type_test.h"
+#include "core/data_type/data_type_factory.hpp"
 #include "core/data_type/data_type_string.h"
 #include "core/data_type_serde/data_type_serde.h"
 #include "core/field.h"
 #include "core/string_buffer.hpp"
 #include "core/string_view.h"
 #include "core/types.h"
+#include "storage/olap_common.h"
 #include "util/mysql_row_buffer.h"
 
 namespace doris {
@@ -257,9 +259,9 @@ TEST_F(DataTypeVarbinaryTest, SerDeWriteColumnToMysql) {
     EXPECT_GT(rb_bin.length(), 0);
 }
 
-TEST_F(DataTypeVarbinaryTest, GetStorageFieldTypeThrows) {
+TEST_F(DataTypeVarbinaryTest, GetStorageFieldType) {
     DataTypeVarbinary dt;
-    EXPECT_THROW({ (void)dt.get_storage_field_type(); }, doris::Exception);
+    EXPECT_THROW(dt.get_storage_field_type(), doris::Exception);
 }
 
 TEST_F(DataTypeVarbinaryTest, GetFieldFromTExprNodeWithEmbeddedNull) {
@@ -285,6 +287,16 @@ TEST_F(DataTypeVarbinaryTest, ToProtobufDefaultLen) {
     EXPECT_EQ(scalar.len(), -1);
 }
 
+TEST_F(DataTypeVarbinaryTest, ProtobufPreservesDeclaredLength) {
+    PTypeDesc type;
+    auto* node = type.add_types();
+    node->set_type(TTypeNodeType::SCALAR);
+    node->mutable_scalar_type()->set_type(TPrimitiveType::VARBINARY);
+    node->mutable_scalar_type()->set_len(2);
+    auto restored = DataTypeFactory::instance().create_data_type(type, false);
+    EXPECT_EQ(assert_cast<const DataTypeVarbinary&>(*restored).len(), 2);
+}
+
 TEST_F(DataTypeVarbinaryTest, GetFieldWithDataTypeNonInline) {
     DataTypeVarbinary dt;
     auto col = dt.create_column();
@@ -299,4 +311,4 @@ TEST_F(DataTypeVarbinaryTest, GetFieldWithDataTypeNonInline) {
     ASSERT_EQ(memcmp(sv.data(), big.data(), sv.size()), 0);
 }
 
-} // namespace doris
\ No newline at end of file
+} // namespace doris
diff --git a/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp b/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp
index 5f03f363c860eb..4a42e7697668ce 100644
--- a/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp
+++ b/be/test/core/data_type_serde/data_type_serde_varbinary_test.cpp
@@ -27,6 +27,7 @@
 #include <memory>
 #include <orc/OrcFile.hh>
 #include <string>
+#include <utility>
 #include <vector>
 
 #include "core/arena.h"
@@ -57,6 +58,74 @@ static std::string make_bytes(size_t n, uint8_t seed = 0x31) {
 
 class DataTypeVarbinarySerDeTest : public ::testing::Test {};
 
+TEST_F(DataTypeVarbinarySerDeTest, HiveTextBinaryUsesBase64InsteadOfJsonBytes) {
+    DataTypeVarbinarySerDe serde;
+    auto column = ColumnVarbinary::create();
+    auto options = DataTypeSerDe::get_default_format_options();
+    // Hive LazyBinary decodes Base64, whereas JSON/CSV and binary file readers keep raw bytes.
+    const std::vector<std::pair<std::string, std::string>> cases = {
+            {"dGVzdDI=", "test2"},
+            {"AP8=", std::string("\0\xff", 2)},
+            {"", ""},
+            {"not!base64", "not!base64"},
+            {"====", "===="},
+            {"dGVzdDI", "test2"},
+            {"dG Vz\tdDI=\r\n", "test2"},
+            {"-_8=", std::string("\xfb\xff", 2)},
+            {"YWJjZ", "abc"}};
+    for (const auto& [encoded, expected] : cases) {
+        Slice slice(encoded);
+        ASSERT_TRUE(serde.deserialize_one_cell_from_hive_text(*column, slice, options).ok());
+        EXPECT_EQ(expected, column->get_data_at(column->size() - 1).to_string());
+    }
+    auto output = ColumnString::create();
+    VectorBufferWriter writer(*output);
+    auto binary = ColumnVarbinary::create();
+    binary->insert_data("\0\xff", 2);
+    ASSERT_TRUE(serde.serialize_one_cell_to_hive_text(*binary, 0, writer, options).ok());
+    writer.commit();
+    EXPECT_EQ("AP8=", output->get_data_at(0).to_string());
+
+    std::string encoded = "dGVzdDI=";
+    Slice raw(encoded);
+    ASSERT_TRUE(serde.deserialize_one_cell_from_json(*column, raw, options).ok());
+    EXPECT_EQ(encoded, column->get_data_at(column->size() - 1).to_string());
+
+    auto vector_column = ColumnVarbinary::create();
+    std::vector<Slice> slices;
+    for (const auto& [text, expected] : cases) {
+        slices.emplace_back(text);
+    }
+    uint64_t count = 0;
+    ASSERT_TRUE(serde.deserialize_column_from_hive_text_vector(*vector_column, slices, &count,
+                                                               options, 2)
+                        .ok());
+    ASSERT_EQ(cases.size(), count);
+    for (size_t i = 0; i < cases.size(); ++i) {
+        EXPECT_EQ(cases[i].second, vector_column->get_data_at(i).to_string());
+    }
+}
+
+TEST_F(DataTypeVarbinarySerDeTest, FromHexStringPreservesBinaryPartitionBytes) {
+    DataTypeVarbinarySerDe serde;
+    auto column = ColumnVarbinary::create();
+    auto options = DataTypeSerDe::get_default_format_options();
+    for (const std::string text : {"0x00FF", "0x", "0x123E4567E89B12D3A456426614174000"}) {
+        StringRef input(text);
+        ASSERT_TRUE(serde.from_string(input, *column, options).ok());
+    }
+    ASSERT_EQ(3, column->size());
+    EXPECT_EQ(std::string("\0\xff", 2), column->get_data_at(0).to_string());
+    EXPECT_EQ(0, column->get_data_at(1).size);
+    EXPECT_EQ(std::string("\x12\x3e\x45\x67\xe8\x9b\x12\xd3\xa4\x56\x42\x66\x14\x17\x40\x00", 16),
+              column->get_data_at(2).to_string());
+    for (const std::string text : {"0x0", "0xGG", "1234"}) {
+        StringRef input(text);
+        EXPECT_FALSE(serde.from_string(input, *column, options).ok());
+        EXPECT_EQ(3, column->size());
+    }
+}
+
 TEST_F(DataTypeVarbinarySerDeTest, Name) {
     DataTypeVarbinarySerDe serde;
     EXPECT_EQ(serde.get_name(), std::string("Varbinary"));
diff --git a/be/test/exec/common/hash_table/hash_key_type_test.cpp b/be/test/exec/common/hash_table/hash_key_type_test.cpp
index 4d68ff70801ee5..770f3433f1df28 100644
--- a/be/test/exec/common/hash_table/hash_key_type_test.cpp
+++ b/be/test/exec/common/hash_table/hash_key_type_test.cpp
@@ -24,9 +24,19 @@
 #include "core/data_type/data_type_nullable.h"
 #include "core/data_type/data_type_number.h"
 #include "core/data_type/data_type_struct.h"
+#include "core/data_type/data_type_varbinary.h"
 
 namespace doris {
 
+TEST(HashKeyTypeTest, BinaryKeysAreNotSupported) {
+    auto type = std::make_shared<DataTypeVarbinary>();
+    for (const auto& key : DataTypes {type, make_nullable(type)}) {
+        EXPECT_THROW(get_hash_key_type({key}), Exception);
+        EXPECT_THROW(get_hash_key_type({key, std::make_shared<DataTypeInt32>()}), Exception);
+        EXPECT_THROW(get_hash_key_type({std::make_shared<DataTypeInt32>(), key}), Exception);
+    }
+}
+
 TEST(HashKeyTypeTest, FixedWidthStructUsesSerializedKey) {
     const auto group_key = make_nullable(std::make_shared<DataTypeInt32>());
 
diff --git a/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp b/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp
index 974eb817e8872a..60d1cc976ae6b0 100644
--- a/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp
+++ b/be/test/exec/sink/writer/iceberg/partition_transformers_test.cpp
@@ -22,6 +22,8 @@
 #include <limits>
 
 #include "core/data_type/data_type_date_or_datetime_v2.h"
+#include "core/data_type/data_type_varbinary.h"
+#include "format/table/iceberg/partition_spec.h"
 
 namespace doris {
 
@@ -31,6 +33,18 @@ class PartitionTransformersTest : public testing::Test {
     virtual ~PartitionTransformersTest() = default;
 };
 
+TEST_F(PartitionTransformersTest, binary_computation_transforms_are_not_supported) {
+    const auto type = std::make_shared<DataTypeVarbinary>();
+    for (const auto& source_type : DataTypes {type, make_nullable(type)}) {
+        for (const auto& transform : {"truncate[1]", "bucket[16]"}) {
+            EXPECT_THROW(
+                    PartitionColumnTransforms::create(
+                            iceberg::PartitionField(1, 1000, "binary_key", transform), source_type),
+                    Exception);
+        }
+    }
+}
+
 TEST_F(PartitionTransformersTest, test_integer_truncate_transform) {
     const std::vector<int32_t> values({1, -1});
     auto column = ColumnInt32::create();
diff --git a/be/test/exprs/aggregate/agg_min_max_test.cpp b/be/test/exprs/aggregate/agg_min_max_test.cpp
index 86b19d7462bad8..c02cbd78841cbe 100644
--- a/be/test/exprs/aggregate/agg_min_max_test.cpp
+++ b/be/test/exprs/aggregate/agg_min_max_test.cpp
@@ -36,6 +36,7 @@
 #include "core/data_type/data_type_number.h"
 #include "core/data_type/data_type_string.h"
 #include "core/data_type/data_type_timestamp_ns.h"
+#include "core/data_type/data_type_varbinary.h"
 #include "core/field.h"
 #include "core/string_ref.h"
 #include "core/types.h"
@@ -49,6 +50,17 @@ namespace doris {
 // declare function
 void register_aggregate_function_minmax(AggregateFunctionSimpleFactory& factory);
 
+TEST(BinaryAggregateTest, SingleValueAggregatesAreNotSupported) {
+    AggregateFunctionSimpleFactory factory;
+    register_aggregate_function_minmax(factory);
+    for (const auto& type : DataTypes {std::make_shared<DataTypeVarbinary>(),
+                                       make_nullable(std::make_shared<DataTypeVarbinary>())}) {
+        for (const auto& name : {"min", "max"}) {
+            EXPECT_THROW(factory.get(name, {type}, nullptr, type->is_nullable(), -1), Exception);
+        }
+    }
+}
+
 class AggMinMaxTest : public ::testing::TestWithParam<std::string> {};
 
 TEST_P(AggMinMaxTest, min_max_test) {
diff --git a/be/test/exprs/expr_zonemap_filter_test.cpp b/be/test/exprs/expr_zonemap_filter_test.cpp
index 88dee1cb9748d5..27dcfdd83440b2 100644
--- a/be/test/exprs/expr_zonemap_filter_test.cpp
+++ b/be/test/exprs/expr_zonemap_filter_test.cpp
@@ -40,6 +40,7 @@
 #include "core/data_type/data_type_string.h"
 #include "core/data_type/data_type_struct.h"
 #include "core/data_type/data_type_time.h"
+#include "core/data_type/data_type_varbinary.h"
 #include "core/field.h"
 #include "core/string_ref.h"
 #include "core/value/vdatetime_value.h"
@@ -76,6 +77,7 @@
 #endif
 
 namespace doris {
+
 namespace {
 
 Field int_field(int32_t value) {
@@ -1507,6 +1509,21 @@ TEST(ExprZonemapFilterTest, VInPredicateDictionaryAndBloomProbePreparedSet) {
               in_predicate->evaluate_bloom_filter(matching_bloom_ctx));
 }
 
+TEST(ExprZonemapFilterTest, BinaryInDoesNotMaterializeStoragePredicates) {
+    auto type = std::make_shared<DataTypeVarbinary>();
+    for (bool negative : {false, true}) {
+        auto predicate = std::make_shared<VInPredicate>(make_in_predicate_node(negative, 2));
+        predicate->add_child(make_slot(0, type));
+        auto field = Field::create_field<TYPE_VARBINARY>(StringView("\0\xff", 2));
+        predicate->add_child(
+                std::make_shared<VLiteral>(create_texpr_node_from(field, TYPE_VARBINARY, 0, 0)));
+        ASSERT_NO_THROW(predicate->_prepare_zonemap_min_max(nullptr));
+        EXPECT_FALSE(predicate->can_evaluate_zonemap_filter());
+        EXPECT_FALSE(predicate->can_evaluate_dictionary_filter());
+        EXPECT_FALSE(predicate->can_evaluate_bloom_filter());
+    }
+}
+
 TEST(ExprZonemapFilterTest, VInPredicatePreparesNestedBloomValuesDuringOpen) {
     auto leaf_type = int_type();
     auto struct_type = std::make_shared<DataTypeStruct>(DataTypes {leaf_type}, Strings {"value"});
diff --git a/be/test/exprs/function/cast/cast_to_string_api_test.cpp b/be/test/exprs/function/cast/cast_to_string_api_test.cpp
index 537090699b163a..2d59760824c659 100644
--- a/be/test/exprs/function/cast/cast_to_string_api_test.cpp
+++ b/be/test/exprs/function/cast/cast_to_string_api_test.cpp
@@ -17,6 +17,8 @@
 
 #include <gtest/gtest.h>
 
+#include "core/data_type/data_type_string.h"
+#include "core/data_type/data_type_timestamptz.h"
 #include "core/types.h"
 #include "core/value/ipv4_value.h"
 #include "core/value/vdatetime_value.h"
@@ -27,6 +29,50 @@
 
 namespace doris {
 
+TEST(CastToStringTest, NullableTimestampSkipsInvalidPayload) {
+    TimestampTzValue valid;
+    valid.unchecked_set_time(2024, 1, 2, 3, 4, 5, 123456);
+    TimestampTzValue invalid;
+    // A NULL row may retain arbitrary bytes from an earlier expression's allocation.
+    invalid.unchecked_set_time(64251, 1, 1, 0, 0, 0);
+    auto input = ColumnTimeStampTz::create();
+    input->insert_value(valid);
+    input->insert_value(invalid);
+    input->insert_value(valid);
+    input->insert_value(invalid);
+    const NullMap null_map {0, 1, 0, 1};
+    ColumnPtr input_column = std::move(input);
+
+    for (size_t rows : {0, 1, 2, 3, 4}) {
+        Block block {{input_column, std::make_shared<DataTypeTimeStampTz>(6), "input"},
+                     {nullptr, std::make_shared<DataTypeString>(), "result"}};
+        ASSERT_NO_THROW({
+            ASSERT_TRUE(CastToStringFunction::execute_impl(nullptr, block, {0}, 1, rows,
+                                                           null_map.data())
+                                .ok());
+        });
+        const auto& result = assert_cast<const ColumnString&>(*block.get_by_position(1).column);
+        ASSERT_EQ(result.size(), rows);
+        for (size_t row = 0; row < rows; ++row) {
+            EXPECT_EQ(result.get_data_at(row).to_string(),
+                      null_map[row] ? "" : "2024-01-02 03:04:05.123456+00:00");
+        }
+    }
+}
+
+TEST(CastToStringTest, NonNullInvalidTimestampIsStillRejected) {
+    TimestampTzValue invalid;
+    invalid.unchecked_set_time(64251, 1, 1, 0, 0, 0);
+    auto input = ColumnTimeStampTz::create();
+    input->insert_value(invalid);
+    Block block {{std::move(input), std::make_shared<DataTypeTimeStampTz>(6), "input"},
+                 {nullptr, std::make_shared<DataTypeString>(), "result"}};
+    const NullMap null_map {0};
+    EXPECT_THROW(static_cast<void>(CastToStringFunction::execute_impl(nullptr, block, {0}, 1, 1,
+                                                                      null_map.data())),
+                 Exception);
+}
+
 TEST(CastToStringTest, test) {
     {
         UInt8 num = 1;
diff --git a/be/test/exprs/function/cast/cast_to_timestamptz_test.cpp b/be/test/exprs/function/cast/cast_to_timestamptz_test.cpp
index cc08b4a69e2549..ee925e95c537d0 100644
--- a/be/test/exprs/function/cast/cast_to_timestamptz_test.cpp
+++ b/be/test/exprs/function/cast/cast_to_timestamptz_test.cpp
@@ -32,6 +32,7 @@
 #include "exprs/function/cast/cast_to_date.h"
 #include "exprs/function/cast/cast_to_timestamp_ns.h"
 #include "exprs/function/cast/cast_wrapper_decls.h"
+#include "exprs/function/functions_comparison.h"
 #include "testutil/column_helper.h"
 #include "testutil/datetime_ut_util.h"
 #include "testutil/mock/mock_runtime_state.h"
@@ -459,4 +460,82 @@ TEST_F(CastTimeStampTzTest, from_timestamptz_non_strict_mode_to_datetime) {
     }
 }
 
+TEST_F(CastTimeStampTzTest, boundary_cast_errors_preserve_status_and_null_semantics) {
+    const auto maximum = make_timestamptz(9999, 12, 31, 23, 59, 59, 999999);
+    for (const bool to_datetime : {false, true}) {
+        auto make_block = [&]() {
+            auto block = ColumnHelper::create_block<DataTypeTimeStampTz>({maximum});
+            block.get_by_position(0).type = std::make_shared<DataTypeTimeStampTz>(6);
+            DataTypePtr target = to_datetime
+                                         ? DataTypePtr(std::make_shared<DataTypeDateTimeV2>(6))
+                                         : DataTypePtr(std::make_shared<DataTypeTimeStampTz>(0));
+            block.insert(ColumnWithTypeAndName {nullptr, target, "result"});
+            return block;
+        };
+        auto strict_block = make_block();
+        Status status;
+        // Local display overflow must not replace the cast's error status with an exception.
+        if (to_datetime) {
+            CastToImpl<CastModeType::StrictMode, DataTypeTimeStampTz, DataTypeDateTimeV2> cast;
+            ASSERT_NO_THROW(
+                    status = cast.execute_impl(&context, strict_block, arguments, result, 1));
+        } else {
+            CastToImpl<CastModeType::StrictMode, DataTypeTimeStampTz, DataTypeTimeStampTz> cast;
+            ASSERT_NO_THROW(
+                    status = cast.execute_impl(&context, strict_block, arguments, result, 1));
+        }
+        // TRY_CAST must recognize a conversion failure instead of propagating an execution error.
+        EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT);
+        EXPECT_NE(status.to_string().find("9999-12-31 23:59:59.999999"), std::string::npos);
+
+        auto nullable_block = make_block();
+        nullable_block.get_by_position(result).type =
+                make_nullable(nullable_block.get_by_position(result).type);
+        if (to_datetime) {
+            CastToImpl<CastModeType::NonStrictMode, DataTypeTimeStampTz, DataTypeDateTimeV2> cast;
+            status = cast.execute_impl(&context, nullable_block, arguments, result, 1, nullptr);
+        } else {
+            CastToImpl<CastModeType::NonStrictMode, DataTypeTimeStampTz, DataTypeTimeStampTz> cast;
+            status = cast.execute_impl(&context, nullable_block, arguments, result, 1, nullptr);
+        }
+        ASSERT_TRUE(status.ok()) << status;
+        const auto& nullable =
+                assert_cast<const ColumnNullable&>(*nullable_block.get_by_position(result).column);
+        EXPECT_TRUE(nullable.get_null_map_data()[0]);
+    }
+}
+
+// NOLINTNEXTLINE(readability-function-cognitive-complexity): GTest exception macros add branches.
+TEST_F(CastTimeStampTzTest, timestamp_ns_local_year_overflow_returns_status) {
+    for (const bool upper : {false, true}) {
+        _state._timezone_obj = cctz::fixed_time_zone(std::chrono::hours(upper ? 8 : -8));
+        const auto value = upper ? make_timestamptz(9999, 12, 31, 23, 59, 59, 999999)
+                                 : make_timestamptz(0, 1, 1, 0, 0, 0, 0);
+        auto block = ColumnHelper::create_block<DataTypeTimeStampTz>({value});
+        block.get_by_position(0).type = std::make_shared<DataTypeTimeStampTz>(6);
+        block.insert({nullptr, std::make_shared<DataTypeTimeStampNs>(), "result"});
+        CastToImpl<CastModeType::StrictMode, DataTypeTimeStampTz, DataTypeTimeStampNs> cast;
+        Status status;
+        ASSERT_NO_THROW(status = cast.execute_impl(&context, block, {0}, 1, 1));
+        EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT);
+        EXPECT_NE(status.to_string().find("can not cast timestamptz"), std::string::npos);
+
+        CastToImpl<CastModeType::NonStrictMode, DataTypeTimeStampTz, DataTypeTimeStampNs> try_cast;
+        ASSERT_TRUE(try_cast.execute_impl(&context, block, {0}, 1, 1).ok());
+        EXPECT_TRUE(block.get_by_position(1).column->is_null_at(0));
+
+        auto ns_column = ColumnTimeStampNs::create();
+        ns_column->insert_default();
+        block.get_by_position(1).column = std::move(ns_column);
+        block.insert({nullptr, std::make_shared<DataTypeUInt8>(), "comparison"});
+        FunctionComparison<EqualsOp, NameEquals> equals;
+        // Error reporting must not try to display the unrepresentable session-local year.
+        for (const ColumnNumbers& inputs : {ColumnNumbers {0, 1}, ColumnNumbers {1, 0}}) {
+            ASSERT_NO_THROW(status = equals.execute_impl(&context, block, inputs, 2, 1));
+            EXPECT_EQ(status.code(), ErrorCode::INVALID_ARGUMENT);
+            EXPECT_NE(status.to_string().find("can not compare timestamptz"), std::string::npos);
+        }
+    }
+}
+
 } // namespace doris
diff --git a/be/test/exprs/function/function_varbinary_test.cpp b/be/test/exprs/function/function_varbinary_test.cpp
index c0fe3b02f1ce13..3de706661ec1d2 100644
--- a/be/test/exprs/function/function_varbinary_test.cpp
+++ b/be/test/exprs/function/function_varbinary_test.cpp
@@ -19,11 +19,26 @@
 
 #include "core/data_type/data_type_varbinary.h"
 #include "exprs/function/function_test_util.h"
+#include "exprs/function/in.h"
 
 namespace doris {
 
 using namespace ut_type;
 
+TEST(function_binary_test, in_and_not_in_are_not_supported) {
+    RuntimeState state;
+    for (const auto& type : DataTypes {std::make_shared<DataTypeVarbinary>(),
+                                       make_nullable(std::make_shared<DataTypeVarbinary>())}) {
+        const DataTypes arguments {type, type};
+        auto context = FunctionContext::create_context(
+                &state, make_nullable(std::make_shared<DataTypeUInt8>()), arguments);
+        EXPECT_EQ(FunctionIn<false>().open(context.get(), FunctionContext::FRAGMENT_LOCAL).code(),
+                  ErrorCode::NOT_IMPLEMENTED_ERROR);
+        EXPECT_EQ(FunctionIn<true>().open(context.get(), FunctionContext::FRAGMENT_LOCAL).code(),
+                  ErrorCode::NOT_IMPLEMENTED_ERROR);
+    }
+}
+
 TEST(function_binary_test, function_binary_length_test) {
     std::string func_name = "length";
     InputTypeSet input_types = {PrimitiveType::TYPE_VARBINARY};
diff --git a/be/test/exprs/vexpr_test.cpp b/be/test/exprs/vexpr_test.cpp
index ddcc5234e0f2e6..10bdfb815d7d57 100644
--- a/be/test/exprs/vexpr_test.cpp
+++ b/be/test/exprs/vexpr_test.cpp
@@ -725,7 +725,9 @@ TEST(TEST_VEXPR, LITERALTEST) {
 
             ColumnPtr result_column;
             ASSERT_TRUE(literal.execute_column(nullptr, nullptr, nullptr, 1, result_column).ok());
-            auto sv = (*result_column)[0].get<TYPE_VARBINARY>();
+            // The view borrows the Field's owned bytes, so keep the Field alive for the assertion.
+            const auto result_field = (*result_column)[0];
+            const auto& sv = result_field.get<TYPE_VARBINARY>();
             EXPECT_EQ(value, std::string(sv.data(), sv.size()));
         }
     }
diff --git a/be/test/runtime/timestamptz_value_test.cpp b/be/test/runtime/timestamptz_value_test.cpp
index ad35e8681c363a..35a3e6ec594211 100644
--- a/be/test/runtime/timestamptz_value_test.cpp
+++ b/be/test/runtime/timestamptz_value_test.cpp
@@ -21,9 +21,13 @@
 #include <cctz/time_zone.h>
 #include <gtest/gtest.h>
 
+#include <chrono>
 #include <string>
+#include <utility>
 
+#include "common/exception.h"
 #include "exprs/function/cast/cast_base.h"
+#include "exprs/function/cast/cast_to_timestamptz_impl.hpp"
 #include "testutil/datetime_ut_util.h"
 #include "util/timezone_utils.h"
 
@@ -34,6 +38,98 @@ TEST(TimeStampTzValueTest, make_time) {
     EXPECT_EQ(tz.to_date_int_val(), MIN_DATETIME_V2);
 }
 
+TEST(TimeStampTzValueTest, ToStringPreservesHistoricalOffsetSeconds) {
+    TimezoneUtils::load_offsets_to_cache();
+    const auto utc = cctz::utc_time_zone();
+    cctz::time_zone shanghai;
+    cctz::time_zone new_york;
+    cctz::time_zone kathmandu;
+    ASSERT_TRUE(cctz::load_time_zone("Asia/Shanghai", &shanghai));
+    ASSERT_TRUE(cctz::load_time_zone("America/New_York", &new_york));
+    ASSERT_TRUE(cctz::load_time_zone("Asia/Kathmandu", &kathmandu));
+    struct TestCase {
+        cctz::time_zone zone;
+        int year;
+        const char* civil;
+        const char* offset;
+    };
+    const TestCase cases[] = {
+            {.zone = shanghai, .year = 1890, .civil = "1890-01-01 08:05:43", .offset = "+08:05:43"},
+            {.zone = new_york, .year = 1880, .civil = "1879-12-31 19:03:58", .offset = "-04:56:02"},
+            // Pre-standard offsets vary across tzdata versions. Fixed zones keep coverage
+            // of offsets beyond 14 hours independent of the host's historical records.
+            {.zone = cctz::fixed_time_zone(std::chrono::seconds(-57368)),
+             .year = 1800,
+             .civil = "1799-12-31 08:03:52",
+             .offset = "-15:56:08"},
+            {.zone = cctz::fixed_time_zone(std::chrono::seconds(-51660)),
+             .year = 1800,
+             .civil = "1799-12-31 09:39:00",
+             .offset = "-14:21"},
+            {.zone = shanghai, .year = 2024, .civil = "2024-01-01 08:00:00", .offset = "+08:00"},
+            {.zone = new_york, .year = 2024, .civil = "2023-12-31 19:00:00", .offset = "-05:00"},
+            {.zone = kathmandu, .year = 2024, .civil = "2024-01-01 05:45:00", .offset = "+05:45"},
+            {.zone = utc, .year = 2024, .civil = "2024-01-01 00:00:00", .offset = "+00:00"},
+    };
+    for (const auto& test_case : cases) {
+        const auto& zone = test_case.zone;
+        for (const auto scale : {0, 3, 6}) {
+            SCOPED_TRACE(testing::Message() << zone.name() << ", scale=" << scale);
+            const auto micros = scale == 6 ? 123456 : scale == 3 ? 123000 : 0;
+            const auto value = make_timestamptz(test_case.year, 1, 1, 0, 0, 0, micros);
+            const std::string fraction = scale == 6 ? ".123456" : scale == 3 ? ".123" : "";
+            const auto formatted = value.to_string(zone, scale);
+            EXPECT_EQ(formatted, std::string(test_case.civil) + fraction + test_case.offset);
+
+            // The client-visible offset must describe the same instant, including historical
+            // sub-minute offsets; parsing in UTC must not depend on the display session zone.
+            for (const bool strict : {false, true}) {
+                TimestampTzValue parsed;
+                CastParameters params;
+                params.is_strict = strict;
+                ASSERT_TRUE(parsed.from_string(StringRef(formatted), &utc, params, scale))
+                        << params.status.to_string();
+                EXPECT_EQ(parsed, value) << formatted;
+            }
+        }
+    }
+}
+
+TEST(TimeStampTzValueTest, ToStringRejectsUnrepresentableLocalYear) {
+    TimezoneUtils::load_offsets_to_cache();
+    const auto utc = cctz::utc_time_zone();
+    const auto east = cctz::fixed_time_zone(std::chrono::hours(8));
+    const auto west = cctz::fixed_time_zone(std::chrono::hours(-8));
+    for (const auto scale : {0, 3, 6}) {
+        const auto micros = scale == 6 ? 999999 : scale == 3 ? 999000 : 0;
+        const auto minimum = make_timestamptz(0, 1, 1, 0, 0, 0, 0);
+        const auto maximum = make_timestamptz(9999, 12, 31, 23, 59, 59, micros);
+        // A valid UTC instant must not turn into an offset-only protocol value.
+        for (const auto& entry : {std::make_pair(minimum, west), std::make_pair(maximum, east)}) {
+            try {
+                static_cast<void>(entry.first.to_string(entry.second, scale));
+                FAIL() << "Expected an unrepresentable local year error";
+            } catch (const Exception& e) {
+                EXPECT_EQ(e.code(), ErrorCode::INVALID_ARGUMENT);
+                EXPECT_NE(std::string(e.what()).find("TIMESTAMPTZ local year is outside [0, 9999]"),
+                          std::string::npos);
+            }
+        }
+        for (const auto& entry : {std::make_pair(minimum, utc), std::make_pair(maximum, utc),
+                                  std::make_pair(minimum, east), std::make_pair(maximum, west)}) {
+            const auto wire = entry.first.to_string(entry.second, scale);
+            for (const bool strict : {false, true}) {
+                TimestampTzValue parsed;
+                CastParameters params;
+                params.is_strict = strict;
+                ASSERT_TRUE(parsed.from_string(StringRef(wire), &utc, params, scale))
+                        << wire << ": " << params.status.to_string();
+                EXPECT_EQ(parsed, entry.first);
+            }
+        }
+    }
+}
+
 TEST(TimeStampTzValueTest, from_string) {
     cctz::time_zone time_zone = cctz::fixed_time_zone(std::chrono::hours(8));
     TimezoneUtils::load_offsets_to_cache();
@@ -110,6 +206,43 @@ TEST(TimeStampTzValueTest, from_string) {
     }
 }
 
+TEST(TimeStampTzValueTest, HistoricalOffsetsInStrictAndFallbackParsers) {
+    const auto utc = cctz::utc_time_zone();
+    const auto expected = make_timestamptz(1890, 1, 1, 0, 0, 0, 123456);
+    for (const std::string input :
+         {"1890-01-01 08:05:43.123456+08:05:43", "1889-12-31 19:03:58.123456-04:56:02",
+          "1890-01-01 00:00:30.123456+00:00:30", "1889-12-31 23:59:30.123456-00:00:30",
+          "1890-01-01 08:05:00.123456+08:05", "1889-12-31 08:03:52.123456-15:56:08",
+          "1889-12-31 09:39:00.123456-14:21", "1890-01-01 15:00:00.123456+15:00",
+          "1889-12-31 11:59:59.123456-12:00:01"}) {
+        SCOPED_TRACE(input);
+        for (const bool fallback : {false, true}) {
+            TimestampTzValue parsed;
+            CastParameters params;
+            params.is_strict = !fallback;
+            const bool success =
+                    fallback
+                            ? CastToTimestampTz::from_string_non_strict_mode_impl(
+                                      StringRef(input), parsed, params, &utc, 6)
+                            : CastToTimestampTz::from_string_strict_mode<DatelikeParseMode::STRICT>(
+                                      StringRef(input), parsed, params, &utc, 6);
+            EXPECT_TRUE(success) << params.status.to_string();
+            EXPECT_EQ(parsed, expected);
+        }
+    }
+    for (const std::string offset : {"+08:60:00", "+08:05:60", "+08:05:", "+08:05:4", "+08:05:430",
+                                     "+24:00:00", "-24:00:00", "+99:00:00"}) {
+        SCOPED_TRACE(offset);
+        const auto input = "1890-01-01 00:00:00" + offset;
+        for (const bool strict : {false, true}) {
+            TimestampTzValue parsed;
+            CastParameters params;
+            params.is_strict = strict;
+            EXPECT_FALSE(parsed.from_string(StringRef(input), &utc, params, 6));
+        }
+    }
+}
+
 TEST(TimeStampTzValueTest, from_datetime) {
     cctz::time_zone time_zone = cctz::fixed_time_zone(std::chrono::hours(8));
     TimezoneUtils::load_offsets_to_cache();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CollectSet.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CollectSet.java
index f93714aca03dc0..775e800d630bf0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CollectSet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/agg/CollectSet.java
@@ -112,6 +112,12 @@ public Expression resultForEmptyInput() {
 
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        // The BE set kernel cannot hash raw VARBINARY; reject it before implicit casts change its type.
+        for (Expression argument : getArguments()) {
+            if (argument.getDataType().isVarBinaryType()) {
+                throw new AnalysisException("collect_set does not support VARBINARY arguments");
+            }
+        }
         if (arity() == 2 && !getArgument(1).isConstant()) {
             throw new AnalysisException(
                     "collect_set requires second parameter must be a constant: "
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContains.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContains.java
index 7f0089dd96d360..86d94f80756503 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContains.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContains.java
@@ -59,6 +59,11 @@ private ArrayContains(ScalarFunctionParams functionParams) {
         super(functionParams);
     }
 
+    @Override
+    public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
+    }
+
     /**
      * withChildren.
      */
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContainsAll.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContainsAll.java
index 50613ffde50843..b2bd31898cd6bd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContainsAll.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayContainsAll.java
@@ -56,6 +56,11 @@ private ArrayContainsAll(ScalarFunctionParams functionParams) {
         super(functionParams);
     }
 
+    @Override
+    public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
+    }
+
     /**
      * withChildren.
      */
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayDistinct.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayDistinct.java
index 9d957b137a229d..e6f40d46c03e33 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayDistinct.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayDistinct.java
@@ -61,6 +61,7 @@ private ArrayDistinct(ScalarFunctionParams functionParams) {
      */
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         DataType argType = getArgument(0).getDataType();
         if (argType.isArrayType()) {
             DataType itemType = ((ArrayType) argType).getItemType();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayEnumerateUniq.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayEnumerateUniq.java
index beac5aaf0ab474..db21fce3365216 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayEnumerateUniq.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayEnumerateUniq.java
@@ -65,6 +65,7 @@ private ArrayEnumerateUniq(ScalarFunctionParams functionParams) {
      */
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         for (Expression arg : getArguments()) {
             DataType argType = arg.getDataType();
             if (argType.isArrayType()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayExcept.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayExcept.java
index 03fb5b459b5a35..e3d19299416418 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayExcept.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayExcept.java
@@ -54,6 +54,11 @@ private ArrayExcept(ScalarFunctionParams functionParams) {
         super(functionParams);
     }
 
+    @Override
+    public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
+    }
+
     /**
      * withChildren.
      */
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayFunctionUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayFunctionUtils.java
new file mode 100644
index 00000000000000..49cd49d481c2a1
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayFunctionUtils.java
@@ -0,0 +1,42 @@
+// 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.
+
+package org.apache.doris.nereids.trees.expressions.functions.scalar;
+
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.DataType;
+
+/** Argument validation shared by array functions. */
+final class ArrayFunctionUtils {
+    private ArrayFunctionUtils() {
+    }
+
+    static void checkNoVarBinaryArguments(ScalarFunction function) {
+        // Inspect original arguments before coercion can hide unsupported binary comparison/hash inputs.
+        for (Expression argument : function.getArguments()) {
+            DataType type = argument.getDataType();
+            while (type instanceof ArrayType) {
+                type = ((ArrayType) type).getItemType();
+            }
+            if (type.isVarBinaryType()) {
+                throw new AnalysisException(function.getName() + " does not support VARBINARY arguments");
+            }
+        }
+    }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayIntersect.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayIntersect.java
index c48b54305ed9a4..9bd8bcc039b534 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayIntersect.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayIntersect.java
@@ -62,6 +62,7 @@ private ArrayIntersect(ScalarFunctionParams functionParams) {
      */
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         DataType itemType = NullType.INSTANCE;
         for (Expression child : getArguments()) {
             DataType argType = child.getDataType();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayPosition.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayPosition.java
index 490b428b062238..945aee4831e63e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayPosition.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayPosition.java
@@ -77,6 +77,7 @@ public ArrayPosition withChildren(List<Expression> children) {
      */
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         DataType argType = getArgument(0).getDataType();
         if (argType.isArrayType() && ((ArrayType) argType).getItemType().isComplexType()) {
             throw new AnalysisException("array_position does not support complex types: " + toSql());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayRemove.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayRemove.java
index 2610f84c8f7e9b..9977b7efbe1ca9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayRemove.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayRemove.java
@@ -72,6 +72,7 @@ public ArrayRemove withChildren(List<Expression> children) {
 
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         DataType argType = getArgument(0).getDataType();
         if (argType.isArrayType() && (((ArrayType) argType).getItemType().isComplexType()
                     || ((ArrayType) argType).getItemType().isVariantType()
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayUnion.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayUnion.java
index fb4fc04e2cdeba..ecc880972b9c1e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayUnion.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArrayUnion.java
@@ -68,6 +68,7 @@ public ArrayUnion withChildren(List<Expression> children) {
 
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         DataType argType = getArgument(0).getDataType();
         if (argType.isArrayType() && (((ArrayType) argType).getItemType().isComplexType()
                     || ((ArrayType) argType).getItemType().isVariantType()
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArraysOverlap.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArraysOverlap.java
index 2f9402f5444378..50d79d58b10dca 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArraysOverlap.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/ArraysOverlap.java
@@ -68,6 +68,7 @@ public ArraysOverlap withChildren(List<Expression> children) {
 
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         DataType argType = getArgument(0).getDataType();
         if (argType.isArrayType() && (((ArrayType) argType).getItemType().isComplexType()
                     || ((ArrayType) argType).getItemType().isVariantType()
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/CountEqual.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/CountEqual.java
index 20c6a9c208a596..3e97e46ff2a499 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/CountEqual.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/CountEqual.java
@@ -64,6 +64,7 @@ private CountEqual(ScalarFunctionParams functionParams) {
 
     @Override
     public void checkLegalityBeforeTypeCoercion() {
+        ArrayFunctionUtils.checkNoVarBinaryArguments(this);
         DataType argType = getArgument(0).getDataType();
         if (argType.isArrayType() && (((ArrayType) argType).getItemType().isComplexType()
                     || ((ArrayType) argType).getItemType().isVariantType()
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/VarBinaryCollectionLegalityTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/VarBinaryCollectionLegalityTest.java
new file mode 100644
index 00000000000000..a811152f76c261
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/trees/expressions/functions/VarBinaryCollectionLegalityTest.java
@@ -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.
+
+package org.apache.doris.nereids.trees.expressions.functions;
+
+import org.apache.doris.nereids.exceptions.AnalysisException;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.CollectSet;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayContains;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayContainsAll;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayDistinct;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayEnumerateUniq;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayExcept;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayIntersect;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayPosition;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayRemove;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArrayUnion;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.ArraysOverlap;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.CountEqual;
+import org.apache.doris.nereids.trees.expressions.literal.IntegerLiteral;
+import org.apache.doris.nereids.types.ArrayType;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.nereids.types.IntegerType;
+import org.apache.doris.nereids.types.StringType;
+import org.apache.doris.nereids.types.VarBinaryType;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class VarBinaryCollectionLegalityTest {
+    private List<BoundFunction> collections(Expression array, Expression value) {
+        return Arrays.asList(
+                new ArrayContains(array, value), new ArrayPosition(array, value), new CountEqual(array, value),
+                new ArrayDistinct(array), new ArrayRemove(array, value), new ArrayEnumerateUniq(array),
+                new ArrayContainsAll(array, array), new ArraysOverlap(array, array), new ArrayUnion(array, array),
+                new ArrayExcept(array, array), new ArrayIntersect(array, array));
+    }
+
+    private void assertRejectsVarBinary(List<BoundFunction> functions) {
+        Assertions.assertAll(functions.stream().map(function -> () -> {
+            AnalysisException error = Assertions.assertThrows(AnalysisException.class,
+                    function::checkLegalityBeforeTypeCoercion, function.getName());
+            Assertions.assertEquals(function.getName() + " does not support VARBINARY arguments", error.getMessage());
+        }));
+    }
+
+    @Test
+    public void testFunctionsRejectVarBinaryBeforeCoercion() {
+        Expression value = new SlotReference("bytes", VarBinaryType.INSTANCE);
+        Expression array = new SlotReference("items", ArrayType.of(VarBinaryType.INSTANCE));
+        assertRejectsVarBinary(collections(array, value));
+        assertRejectsVarBinary(Arrays.asList(new CollectSet(value), new CollectSet(value, new IntegerLiteral(2))));
+    }
+
+    @Test
+    public void testNestedArraysRejectVarBinary() {
+        Expression value = new SlotReference("bytes", ArrayType.of(VarBinaryType.INSTANCE));
+        Expression array = new SlotReference("items", ArrayType.of(ArrayType.of(VarBinaryType.INSTANCE)));
+        assertRejectsVarBinary(collections(array, value));
+    }
+
+    @Test
+    public void testChecksAllArgumentsBeforeCoercion() {
+        Expression value = new SlotReference("bytes", VarBinaryType.INSTANCE);
+        Expression binaryArray = new SlotReference("bytes_array", ArrayType.of(VarBinaryType.INSTANCE));
+        Expression stringArray = new SlotReference("text_array", ArrayType.of(StringType.INSTANCE));
+        assertRejectsVarBinary(Arrays.asList(
+                new ArrayContains(stringArray, value), new ArrayPosition(stringArray, value),
+                new CountEqual(stringArray, value), new ArrayRemove(stringArray, value),
+                new ArrayContainsAll(stringArray, binaryArray), new ArraysOverlap(stringArray, binaryArray),
+                new ArrayExcept(stringArray, binaryArray), new ArrayUnion(stringArray, stringArray, binaryArray),
+                new ArrayIntersect(stringArray, stringArray, binaryArray),
+                new ArrayEnumerateUniq(stringArray, binaryArray), new CollectSet(new IntegerLiteral(1), value)));
+    }
+
+    @Test
+    public void testOrdinaryTypesRemainLegal() {
+        for (DataType type : Arrays.asList(IntegerType.INSTANCE, StringType.INSTANCE)) {
+            Expression value = new SlotReference("value", type);
+            Expression array = new SlotReference("items", ArrayType.of(type));
+            for (BoundFunction function : collections(array, value)) {
+                Assertions.assertDoesNotThrow(function::checkLegalityBeforeTypeCoercion, function.getName());
+            }
+            Assertions.assertDoesNotThrow(new CollectSet(value)::checkLegalityBeforeTypeCoercion);
+            Assertions.assertDoesNotThrow(new CollectSet(value, new IntegerLiteral(2))::checkLegalityBeforeTypeCoercion);
+        }
+    }
+
+    @Test
+    public void testCollectSetRetainsConstantLimitCheck() {
+        CollectSet function = new CollectSet(new IntegerLiteral(1), new SlotReference("limit", IntegerType.INSTANCE));
+        AnalysisException error = Assertions.assertThrows(AnalysisException.class,
+                function::checkLegalityBeforeTypeCoercion);
+        Assertions.assertTrue(error.getMessage().contains("second parameter must be a constant"));
+    }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/types/VarBinaryUnsupportedCollectionTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/types/VarBinaryUnsupportedCollectionTest.java
new file mode 100644
index 00000000000000..39f5368009d051
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/types/VarBinaryUnsupportedCollectionTest.java
@@ -0,0 +1,62 @@
+// 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.
+
+package org.apache.doris.nereids.types;
+
+import org.apache.doris.nereids.util.PlanChecker;
+import org.apache.doris.utframe.TestWithFeService;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class VarBinaryUnsupportedCollectionTest extends TestWithFeService {
+    @Override
+    protected void runBeforeAll() throws Exception {
+        createDatabaseAndUse("binary_collections");
+        createTable("create table source_bytes (id int, encoded string) duplicate key(id) "
+                + "distributed by hash(id) buckets 1 properties ('replication_num'='1')");
+    }
+
+    @Test
+    public void testUnsupportedBinaryCollectionsFailDuringAnalysis() {
+        String values = "array(cast(encoded as varbinary), X'', X'0080FF', NULL)";
+        for (String expression : new String[] {
+                "array_contains(" + values + ", X'0080FF')",
+                "array_position(" + values + ", X'0080FF')",
+                "countequal(" + values + ", X'0080FF')",
+                "array_distinct(" + values + ")",
+                "array_remove(" + values + ", X'0080FF')",
+                "array_enumerate_uniq(" + values + ")",
+                "array_contains_all(" + values + ", " + values + ")",
+                "arrays_overlap(" + values + ", " + values + ")",
+                "array_union(" + values + ", " + values + ")",
+                "array_except(" + values + ", " + values + ")",
+                "array_intersect(" + values + ", " + values + ")",
+                "collect_set(cast(encoded as varbinary))",
+                "collect_set(cast(encoded as varbinary), 2)"}) {
+            org.apache.doris.nereids.exceptions.AnalysisException error = Assertions.assertThrows(
+                    org.apache.doris.nereids.exceptions.AnalysisException.class,
+                    () -> PlanChecker.from(connectContext).analyze("select " + expression + " from source_bytes"),
+                    expression);
+            Assertions.assertTrue(error.getMessage().contains("does not support VARBINARY"), error.getMessage());
+        }
+        // Byte-agnostic array construction and element access remain supported.
+        PlanChecker.from(connectContext).analyze("select array(cast(encoded as varbinary))[1] from source_bytes");
+        PlanChecker.from(connectContext).analyze("select collect_list(cast(encoded as varbinary)) from source_bytes");
+    }
+
+}
diff --git a/regression-test/data/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.out b/regression-test/data/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.out
index a59b2e04876f74..8c0d0dd2ba25e1 100644
--- a/regression-test/data/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.out
+++ b/regression-test/data/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.out
@@ -3,7 +3,7 @@
 \N	-1
 \N	12
 \N	12
-0000-01-01 08:05:43+08:05	0
+0000-01-01 08:05:43+08:05:43	0
 2023-01-01 17:00:00+08:00	1
 2023-02-02 17:00:00+08:00	2
 2023-03-04 01:00:00+08:00	3
@@ -13,7 +13,7 @@
 
 -- !dup_key_strict0 --
 \N	-1
-0000-01-01 08:05:43+08:05	0
+0000-01-01 08:05:43+08:05:43	0
 2023-01-01 17:00:00+08:00	1
 2023-02-02 17:00:00+08:00	2
 2023-03-04 01:00:00+08:00	3
@@ -26,7 +26,7 @@
 -- !dup_key_null_to_not_null_non_strict0 --
 
 -- !dup_key_null_to_not_null_non_strict1 --
-0000-01-01 08:05:43+08:05	0
+0000-01-01 08:05:43+08:05:43	0
 2023-01-01 17:00:00+08:00	1
 2023-02-02 17:00:00+08:00	2
 2023-03-04 01:00:00+08:00	3
@@ -35,7 +35,7 @@
 2023-12-12 11:12:12+08:00	12
 
 -- !dup_key_null_to_not_null_strict0 --
-0000-01-01 08:05:43+08:05	0
+0000-01-01 08:05:43+08:05:43	0
 2023-01-01 17:00:00+08:00	1
 2023-02-02 17:00:00+08:00	2
 2023-03-04 01:00:00+08:00	3
diff --git a/regression-test/data/datatype_p0/timestamptz/test_cast_timestamptz.out b/regression-test/data/datatype_p0/timestamptz/test_cast_timestamptz.out
index d9021431ae7f6c..c1c674e3416b43 100644
--- a/regression-test/data/datatype_p0/timestamptz/test_cast_timestamptz.out
+++ b/regression-test/data/datatype_p0/timestamptz/test_cast_timestamptz.out
@@ -14,6 +14,12 @@
 -- !cast_str_to_timetz_invalid --
 \N	\N	\N	\N	\N	\N	\N	\N
 
+-- !cast_str_to_timetz_historical_offsets --
+2019-12-31 16:00:00+07:00	1800-01-01 22:56:08+07:00	1800-01-01 21:21:00+07:00
+
+-- !cast_str_to_timetz_historical_offsets --
+2019-12-31 16:00:00+07:00	1800-01-01 22:56:08+07:00	1800-01-01 21:21:00+07:00
+
 -- !sql --
 2020-01-01 00:00:00.124+07:00
 
diff --git a/regression-test/data/datatype_p0/timestamptz/test_timestamptz_binary_output.out b/regression-test/data/datatype_p0/timestamptz/test_timestamptz_binary_output.out
index b40c83ee9232bd..461cd06f347755 100644
--- a/regression-test/data/datatype_p0/timestamptz/test_timestamptz_binary_output.out
+++ b/regression-test/data/datatype_p0/timestamptz/test_timestamptz_binary_output.out
@@ -15,11 +15,11 @@
 
 -- !all_bin0 --
 \N	\N	0
-\N	0000-01-01 08:05:43+08:05	1
+\N	0000-01-01 08:05:43+08:05:43	1
 \N	2023-08-08 20:20:20+08:00	2
 \N	9999-12-31 23:59:59+08:00	-1
-0000-01-01 08:05:43+08:05	0000-01-01 08:05:43+08:05	0
-0000-01-01 08:05:43+08:05	0000-01-01 08:05:43+08:05	1
+0000-01-01 08:05:43+08:05:43	0000-01-01 08:05:43+08:05:43	0
+0000-01-01 08:05:43+08:05:43	0000-01-01 08:05:43+08:05:43	1
 2023-01-01 12:00:00+08:00	2023-01-01 12:00:00+08:00	0
 2023-08-08 20:20:20+08:00	2023-08-08 20:20:20+08:00	1
 2023-12-12 12:12:12+08:00	2023-12-12 12:12:12+08:00	2
@@ -47,11 +47,11 @@
 
 -- !all_bin_scale0 --
 \N	\N	-1
-0000-01-01 08:05:43.000000+08:05	0000-01-01 08:05:43.000000+08:05	0
-0000-01-01 08:05:43.000000+08:05	0000-01-01 08:05:43.000000+08:05	0
-0000-01-01 08:05:43.000001+08:05	0000-01-01 08:05:43.000001+08:05	0
-0000-01-01 08:05:43.123456+08:05	0000-01-01 08:05:43.123456+08:05	0
-0000-01-01 08:05:43.999999+08:05	0000-01-01 08:05:43.999999+08:05	10
+0000-01-01 08:05:43.000000+08:05:43	0000-01-01 08:05:43.000000+08:05:43	0
+0000-01-01 08:05:43.000000+08:05:43	0000-01-01 08:05:43.000000+08:05:43	0
+0000-01-01 08:05:43.000001+08:05:43	0000-01-01 08:05:43.000001+08:05:43	0
+0000-01-01 08:05:43.123456+08:05:43	0000-01-01 08:05:43.123456+08:05:43	0
+0000-01-01 08:05:43.999999+08:05:43	0000-01-01 08:05:43.999999+08:05:43	10
 2023-08-09 04:20:20.000000+08:00	2023-08-09 04:20:20.000000+08:00	8
 2023-08-09 04:20:20.000000+08:00	2023-08-09 04:20:20.000000+08:00	8
 2023-08-09 04:20:20.000001+08:00	2023-08-09 04:20:20.000001+08:00	8
diff --git a/regression-test/suites/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.groovy b/regression-test/suites/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.groovy
index 5bb5fe4d370108..2b2487cfc3996e 100644
--- a/regression-test/suites/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.groovy
+++ b/regression-test/suites/datatype_p0/timestamptz/load/stream_load/test_timestamptz_stream_load.groovy
@@ -16,6 +16,8 @@
 // under the License.
 
 suite("test_timestamptz_stream_load") {
+    // Named zones retain historical second offsets when loaded timestamps are rendered.
+    sql "set time_zone = 'Asia/Shanghai'"
     def csvFile = """test_timestamptz_stream_load.csv"""
     def prepare_table_dup_key = {
         sql """ DROP TABLE IF EXISTS test_timestamptz_stream_load_dup_key"""
diff --git a/regression-test/suites/datatype_p0/timestamptz/test_cast_timestamptz.groovy b/regression-test/suites/datatype_p0/timestamptz/test_cast_timestamptz.groovy
index 7bd92c87588402..660409bc5264d3 100644
--- a/regression-test/suites/datatype_p0/timestamptz/test_cast_timestamptz.groovy
+++ b/regression-test/suites/datatype_p0/timestamptz/test_cast_timestamptz.groovy
@@ -58,6 +58,7 @@ suite("test_cast_timestamptz") {
         cast('2020-12-31 23:59:59' as TIMESTAMPTZ) as ts_no_tz_winter;
     """
 
+    // Wire offsets can exceed session-zone limits; an offset of 24 hours is invalid.
     qt_cast_str_to_timetz_invalid """
     SELECT 
         cast('2020-13-01 00:00:00 +03:00' as TIMESTAMPTZ) as ts_invalid_month,
@@ -65,12 +66,23 @@ suite("test_cast_timestamptz") {
         cast('2020-01-01 24:00:00 +03:00' as TIMESTAMPTZ) as ts_invalid_hour,
         cast('2020-01-01 00:60:00 +03:00' as TIMESTAMPTZ) as ts_invalid_minute,
         cast('2020-01-01 00:00:60 +03:00' as TIMESTAMPTZ) as ts_invalid_second,
-        cast('2020-01-01 00:00:00 +15:00' as TIMESTAMPTZ) as ts_invalid_tz_hour,
+        cast('2020-01-01 00:00:00 +24:00' as TIMESTAMPTZ) as ts_invalid_tz_hour,
         cast('2020-01-01 00:00:00 +03:60' as TIMESTAMPTZ) as ts_invalid_tz_minute,
         cast('invalid-string' as TIMESTAMPTZ) as ts_invalid_string;
     """
 
 
+    // Historical wire offsets must parse in both cast modes, independently of session-zone limits.
+    for (boolean strict : [false, true]) {
+        sql "set enable_strict_cast=${strict}"
+        qt_cast_str_to_timetz_historical_offsets """
+            select cast('2020-01-01 00:00:00+15:00' as timestamptz),
+                   cast('1800-01-01 00:00:00-15:56:08' as timestamptz),
+                   cast('1800-01-01 00:00:00-14:21' as timestamptz);
+        """
+    }
+    sql "set enable_strict_cast=false"
+
     qt_sql """
         select cast(cast("2020-01-01 00:00:00.1236" as datetime(4)) as timestamptz(3));
     """
diff --git a/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_binary_output.groovy b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_binary_output.groovy
index bd7198a7ee8c95..e3186e0a3da9b6 100644
--- a/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_binary_output.groovy
+++ b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_binary_output.groovy
@@ -64,6 +64,8 @@ suite("test_timestamptz_binary_output") {
     String url = getServerPrepareJdbcUrl(context.config.jdbcUrl, "regression_test_datatype_p0_timestamptz");
     logger.info("jdbc prepare statement url: ${url}")
     def result1 = connect(user, password, url) {
+        // A prepared connection has its own session; exercise historical second offsets explicitly.
+        sql "set time_zone = 'Asia/Shanghai'"
         qt_all_bin0 """
             SELECT * FROM test_timestamptz_binary_output_no_scale ORDER BY 1, 2, 3;
         """
@@ -106,6 +108,8 @@ suite("test_timestamptz_binary_output") {
         SELECT * FROM test_timestamptz_binary_output_with_scale ORDER BY 1, 2, 3;
     """
     def result2 = connect(user, password, url) {
+        // Keep the scaled binary-protocol check independent of the server's default time zone.
+        sql "set time_zone = 'Asia/Shanghai'"
         qt_all_bin_scale0 """
             SELECT * FROM test_timestamptz_binary_output_with_scale ORDER BY 1, 2, 3;
         """
diff --git a/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_historical_offset.groovy b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_historical_offset.groovy
new file mode 100644
index 00000000000000..ca9c5161f32e88
--- /dev/null
+++ b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_historical_offset.groovy
@@ -0,0 +1,53 @@
+// 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.
+
+suite("test_timestamptz_historical_offset") {
+    def originalZone = sql("select @@time_zone")[0][0]
+    def originalStrict = sql("select @@enable_strict_cast")[0][0]
+    def cases = [
+        ["Asia/Shanghai", "1890-01-01 00:00:00.123456+00:00", "1890-01-01 08:05:43.123456+08:05:43"],
+        ["America/New_York", "1880-01-01 00:00:00.123456+00:00", "1879-12-31 19:03:58.123456-04:56:02"],
+        ["Asia/Shanghai", "2024-01-01 00:00:00.123456+00:00", "2024-01-01 08:00:00.123456+08:00"],
+        ["America/New_York", "2024-01-01 00:00:00.123456+00:00", "2023-12-31 19:00:00.123456-05:00"],
+        ["Asia/Kathmandu", "2024-01-01 00:00:00.123456+00:00", "2024-01-01 05:45:00.123456+05:45"]
+    ]
+    try {
+        for (def testCase : cases) {
+            sql "set time_zone = '${testCase[0]}'"
+            for (def strict : [false, true]) {
+                sql "set enable_strict_cast = ${strict}"
+                // A nonconstant input exercises BE protocol formatting and parsing instead of
+                // FE constant folding. The offset must retain the instant when sent back by a client.
+                def wire = sql("""
+                    select cast(concat('${testCase[1]}', substring(cast(number as string), 2))
+                                as timestamptz(6))
+                    from numbers('number' = '1')
+                """)[0][0].toString()
+                assertEquals(testCase[2], wire)
+                def roundTrip = sql("""
+                    select cast(concat('${wire}', substring(cast(number as string), 2))
+                                as timestamptz(6)) = cast('${testCase[1]}' as timestamptz(6))
+                    from numbers('number' = '1')
+                """)
+                assertEquals([[true]], roundTrip)
+            }
+        }
+    } finally {
+        sql "set time_zone = '${originalZone}'"
+        sql "set enable_strict_cast = ${originalStrict}"
+    }
+}
diff --git a/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_null_string.groovy b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_null_string.groovy
new file mode 100644
index 00000000000000..201e969a0ed4ae
--- /dev/null
+++ b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_null_string.groovy
@@ -0,0 +1,48 @@
+// 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.
+
+suite("test_timestamptz_null_string") {
+    def originalZone = sql("select @@time_zone")[0][0]
+    def originalStrict = sql("select @@enable_strict_cast")[0][0]
+    def originalSkipFold = sql("select @@debug_skip_fold_constant")[0][0]
+    try {
+        sql "set time_zone = '+08:00'"
+        sql "set enable_strict_cast = false"
+        for (def skipFold : [false, true]) {
+            sql "set debug_skip_fold_constant = ${skipFold}"
+            // Invalid casts and NULL inputs leave no valid timestamp payload for the
+            // following formatter; only the null map determines whether a row is readable.
+            assertEquals([[null]], sql("""
+                select cast(second_floor('9999-12-31 23:59:59.999999-02:00', 5) as string)
+            """))
+            assertEquals([[null]], sql("""
+                select cast(second_floor(cast(null as timestamptz(6)), 5) as string)
+            """))
+            assertEquals([[null], ['2024-01-02 11:04:05.000000+08:00'],
+                          [null], ['2024-01-02 11:04:05.000000+08:00']], sql("""
+                select cast(second_floor(
+                    if(number % 2 = 0, cast(null as timestamptz(6)),
+                       cast('2024-01-02 03:04:05.123456+00:00' as timestamptz(6))), 5) as string)
+                from numbers('number' = '4') order by number
+            """))
+        }
+    } finally {
+        sql "set time_zone = '${originalZone}'"
+        sql "set enable_strict_cast = ${originalStrict}"
+        sql "set debug_skip_fold_constant = ${originalSkipFold}"
+    }
+}
diff --git a/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_output_boundary.groovy b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_output_boundary.groovy
new file mode 100644
index 00000000000000..f1cde7a38de72c
--- /dev/null
+++ b/regression-test/suites/datatype_p0/timestamptz/test_timestamptz_output_boundary.groovy
@@ -0,0 +1,71 @@
+// 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.
+
+suite("test_timestamptz_output_boundary") {
+    def originalZone = sql("select @@time_zone")[0][0]
+    def originalStrict = sql("select @@enable_strict_cast")[0][0]
+    def minimum = "0000-01-01 00:00:00.000000+00:00"
+    def maximum = "9999-12-31 23:59:59.999999+00:00"
+    def readTimestamp = { value ->
+        // Nonconstant inputs reach the BE formatter instead of FE constant folding.
+        """select cast(concat('${value}', substring(cast(number as string), 2)) as timestamptz(6))
+           from numbers('number' = '1')"""
+    }
+    try {
+        for (def strict : [false, true]) {
+            sql "set enable_strict_cast = ${strict}"
+            sql "set time_zone = '+08:00'"
+            for (def target : [["datetimev2(6)", "to datetime in timezone"],
+                               ["timestamptz(0)", "to timestamptz in timezone"]]) {
+                // Failed casts must retain their error/NULL contract even when the input
+                // cannot be displayed in the session timezone while reporting the error.
+                def query = """
+                    select cast(cast(concat('${maximum}', substring(cast(number as string), 2))
+                                     as timestamptz(6)) as ${target[0]})
+                    from numbers('number' = '1')
+                """
+                if (strict) {
+                    test {
+                        sql query
+                        exception target[1]
+                    }
+                } else {
+                    assertEquals([[null]], sql(query))
+                }
+            }
+            for (def entry : [["-08:00", minimum], ["+08:00", maximum]]) {
+                sql "set time_zone = '${entry[0]}'"
+                test {
+                    sql readTimestamp(entry[1])
+                    exception "TIMESTAMPTZ local year is outside [0, 9999]"
+                }
+            }
+            for (def entry : [["UTC", minimum, minimum], ["UTC", maximum, maximum],
+                              ["+08:00", minimum, "0000-01-01 08:00:00.000000+08:00"],
+                              ["-08:00", maximum, "9999-12-31 15:59:59.999999-08:00"]]) {
+                sql "set time_zone = '${entry[0]}'"
+                def wire = sql(readTimestamp(entry[1]))[0][0].toString()
+                assertEquals(entry[2], wire)
+                sql "set time_zone = 'UTC'"
+                assertEquals(entry[1], sql(readTimestamp(wire))[0][0].toString())
+            }
+        }
+    } finally {
+        sql "set time_zone = '${originalZone}'"
+        sql "set enable_strict_cast = ${originalStrict}"
+    }
+}