diff --git a/velox/core/SimpleFunctionMetadata.h b/velox/core/SimpleFunctionMetadata.h
index 0d9cc0f8b4bcd..935887bf85981 100644
--- a/velox/core/SimpleFunctionMetadata.h
+++ b/velox/core/SimpleFunctionMetadata.h
@@ -24,6 +24,7 @@
 #include "velox/core/CoreTypeSystem.h"
 #include "velox/core/Metaprogramming.h"
 #include "velox/core/QueryConfig.h"
+#include "velox/expression/ComplexViewTypes.h"
 #include "velox/expression/FunctionSignature.h"
 #include "velox/expression/SignatureBinder.h"
 #include "velox/type/SimpleFunctionApi.h"
@@ -894,7 +895,7 @@ class UDFHolder {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& config,
-      const typename exec_resolver<TArgs>::in_type*... constantArgs) {
+      const facebook::velox::exec::OptionalAccessor<typename exec_resolver<TArgs>::in_type>*... constantArgs) {
     if constexpr (udf_has_initialize) {
       return instance_.initialize(inputTypes, config, constantArgs...);
     }
diff --git a/velox/examples/SimpleFunctions.cpp b/velox/examples/SimpleFunctions.cpp
index 6d7c5500c2812..0352a95e40f4e 100644
--- a/velox/examples/SimpleFunctions.cpp
+++ b/velox/examples/SimpleFunctions.cpp
@@ -333,8 +333,8 @@ struct MyRegexpMatchFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig&,
-      const arg_type<Varchar>*,
-      const arg_type<Varchar>* pattern) {
+      const accessor_arg_type<Varchar>*,
+      const accessor_arg_type<Varchar>* pattern) {
     // Compile the RE2 object just once if pattern is constant. Functions might
     // choose to throw in case the regexp pattern is not constant, as it can be
     // quite expensive to compile it on a per-row basis. In this example we
diff --git a/velox/expression/SimpleFunctionAdapter.h b/velox/expression/SimpleFunctionAdapter.h
index 2790a3eaaff35..a213d0a6a1545 100644
--- a/velox/expression/SimpleFunctionAdapter.h
+++ b/velox/expression/SimpleFunctionAdapter.h
@@ -227,8 +227,9 @@ class SimpleFunctionAdapter : public VectorFunction {
       if (packed.at(POSITION) != nullptr) {
         SelectivityVector rows(1);
         DecodedVector decodedVector(*packed.at(POSITION), rows);
-        auto oneReader = VectorReader<arg_at<POSITION>>(&decodedVector);
-        auto accessor = OptionalAccessor(&oneReader, 0);
+        const auto oneReader = VectorReader<arg_at<POSITION>>(&decodedVector);
+        using temp_type = exec_arg_at<POSITION>;
+        auto accessor = OptionalAccessor<temp_type>(&oneReader, 0);
         unpackInitialize<POSITION + 1>(
               inputTypes,
               config,
diff --git a/velox/expression/tests/SimpleFunctionInitTest.cpp b/velox/expression/tests/SimpleFunctionInitTest.cpp
index b0ea89621343d..e914930cb2654 100644
--- a/velox/expression/tests/SimpleFunctionInitTest.cpp
+++ b/velox/expression/tests/SimpleFunctionInitTest.cpp
@@ -39,8 +39,8 @@ struct NonDefaultWithArrayInitFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<int32_t>* /*first*/,
-      const arg_type<velox::Array<int32_t>>* second) {
+      const accessor_arg_type<int32_t>* /*first*/,
+      const accessor_arg_type<velox::Array<int32_t>>* second) {
     if (second == nullptr) {
       return;
     }
@@ -134,8 +134,8 @@ struct NonDefaultWithMapInitFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<int32_t>* /*first*/,
-      const arg_type<velox::Map<int32_t, int64_t>>* second) {
+      const accessor_arg_type<int32_t>* /*first*/,
+      const accessor_arg_type<velox::Map<int32_t, int64_t>>* second) {
     if (second == nullptr) {
       return;
     }
@@ -204,7 +204,7 @@ struct InitAlwaysThrowsFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<int32_t>* /*first*/) {
+      const accessor_arg_type<int32_t>* /*first*/) {
     VELOX_USER_FAIL("Unconditional throw!");
   }
 
diff --git a/velox/expression/tests/SimpleFunctionTest.cpp b/velox/expression/tests/SimpleFunctionTest.cpp
index ae5d1b75a7454..5b47f486f174a 100644
--- a/velox/expression/tests/SimpleFunctionTest.cpp
+++ b/velox/expression/tests/SimpleFunctionTest.cpp
@@ -1253,12 +1253,12 @@ struct ConstantArgumentFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<int32_t>* /*first*/,
-      const arg_type<int32_t>* /*second*/,
-      const arg_type<Varchar>* /*third*/,
-      const arg_type<Generic<T1>>* /*fourth*/,
-      const arg_type<Array<int32_t>>* /*fifth*/,
-      const arg_type<Map<int32_t, int32_t>>* /*sixth*/) {}
+      const accessor_arg_type<int32_t>* /*first*/,
+      const accessor_arg_type<int32_t>* /*second*/,
+      const accessor_arg_type<Varchar>* /*third*/,
+      const accessor_arg_type<Generic<T1>>* /*fourth*/,
+      const accessor_arg_type<Array<int32_t>>* /*fifth*/,
+      const accessor_arg_type<Map<int32_t, int32_t>>* /*sixth*/) {}
 
   bool callNullable(
       out_type<int64_t>& out,
@@ -1301,7 +1301,7 @@ struct DecimalPlusOneFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      const A* /*a*/) {
+      const accessor_arg_type<A>* /*a*/) {
     scale_ = getDecimalPrecisionScale(*inputTypes[0]).second;
   }
 
@@ -1322,7 +1322,7 @@ struct DecimalPlusTwoFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      const A* /*a*/) {
+      const accessor_arg_type<A>* /*a*/) {
     scale_ = getDecimalPrecisionScale(*inputTypes[0]).second;
   }
 
diff --git a/velox/functions/lib/Re2Functions.h b/velox/functions/lib/Re2Functions.h
index 5e7267c7ea831..bfa617c47b2a0 100644
--- a/velox/functions/lib/Re2Functions.h
+++ b/velox/functions/lib/Re2Functions.h
@@ -277,9 +277,9 @@ struct Re2RegexpReplace {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* /*string*/,
-      const arg_type<Varchar>* pattern,
-      const arg_type<Varchar>* replacement) {
+      const accessor_arg_type<Varchar>* /*string*/,
+      const accessor_arg_type<Varchar>* pattern,
+      const accessor_arg_type<Varchar>* replacement) {
     if (pattern != nullptr) {
       const auto processedPattern = prepareRegexpPattern(*pattern);
       re_.emplace(processedPattern, RE2::Quiet);
@@ -303,8 +303,8 @@ struct Re2RegexpReplace {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* string,
-      const arg_type<Varchar>* pattern) {
+      const accessor_arg_type<Varchar>* string,
+      const accessor_arg_type<Varchar>* pattern) {
     initialize(inputTypes, config, string, pattern, nullptr);
   }
 
diff --git a/velox/functions/lib/TimeUtils.h b/velox/functions/lib/TimeUtils.h
index 39892124e0a77..d3e61cb94a776 100644
--- a/velox/functions/lib/TimeUtils.h
+++ b/velox/functions/lib/TimeUtils.h
@@ -100,7 +100,7 @@ struct InitSessionTimezone {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Timestamp>* /*timestamp*/) {
+      const accessor_arg_type<Timestamp>* /*timestamp*/) {
     timeZone_ = getTimeZoneFromConfig(config);
   }
 };
diff --git a/velox/functions/prestosql/DateTimeFunctions.h b/velox/functions/prestosql/DateTimeFunctions.h
index 51eb5f03d89f5..96ed075a309d8 100644
--- a/velox/functions/prestosql/DateTimeFunctions.h
+++ b/velox/functions/prestosql/DateTimeFunctions.h
@@ -59,8 +59,8 @@ struct FromUnixtimeFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<double>* /*unixtime*/,
-      const arg_type<Varchar>* timezone) {
+      const accessor_arg_type<double>* /*unixtime*/,
+      const accessor_arg_type<Varchar>* timezone) {
     if (timezone != nullptr) {
       tzID_ = tz::getTimeZoneID((std::string_view)(*timezone));
     }
@@ -80,8 +80,8 @@ struct FromUnixtimeFunction {
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
       const arg_type<double>* /*unixtime*/,
-      const arg_type<int64_t>* hours,
-      const arg_type<int64_t>* minutes) {
+      const accessor_arg_type<int64_t>* hours,
+      const accessor_arg_type<int64_t>* minutes) {
     if (hours != nullptr && minutes != nullptr) {
       tzID_ = tz::getTimeZoneID(*hours * 60 + *minutes);
     }
@@ -145,21 +145,21 @@ struct DateFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* date) {
+      const accessor_arg_type<Varchar>* date) {
     timeZone_ = getTimeZoneFromConfig(config);
   }
 
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Timestamp>* timestamp) {
+      const accessor_arg_type<Timestamp>* timestamp) {
     timeZone_ = getTimeZoneFromConfig(config);
   }
 
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<TimestampWithTimezone>* timestampWithTimezone) {
+      const accessor_arg_type<TimestampWithTimezone>* timestampWithTimezone) {
     // Do nothing. Session timezone doesn't affect the result.
   }
 
@@ -971,8 +971,8 @@ struct DateTruncFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* unitString,
-      const arg_type<Timestamp>* /*timestamp*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<Timestamp>* /*timestamp*/) {
     timeZone_ = getTimeZoneFromConfig(config);
 
     if (unitString != nullptr) {
@@ -983,8 +983,8 @@ struct DateTruncFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* unitString,
-      const arg_type<Date>* /*date*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<Date>* /*date*/) {
     if (unitString != nullptr) {
       unit_ = getDateUnit(*unitString, false);
     }
@@ -993,8 +993,8 @@ struct DateTruncFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* unitString,
-      const arg_type<TimestampWithTimezone>* /*timestamp*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<TimestampWithTimezone>* /*timestamp*/) {
     if (unitString != nullptr) {
       unit_ = getTimestampUnit(*unitString);
     }
@@ -1196,9 +1196,9 @@ struct DateAddFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* unitString,
-      const int64_t* /*value*/,
-      const arg_type<Timestamp>* /*timestamp*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<int64_t>* /*value*/,
+      const accessor_arg_type<Timestamp>* /*timestamp*/) {
     sessionTimeZone_ = getTimeZoneFromConfig(config);
     if (unitString != nullptr) {
       unit_ = fromDateTimeUnitString(*unitString, false /*throwIfInvalid*/);
@@ -1208,9 +1208,9 @@ struct DateAddFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* unitString,
-      const int64_t* /*value*/,
-      const arg_type<Date>* /*date*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<int64_t>* /*value*/,
+      const accessor_arg_type<Date>* /*date*/) {
     if (unitString != nullptr) {
       unit_ = getDateUnit(*unitString, false);
     }
@@ -1295,9 +1295,9 @@ struct DateDiffFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* unitString,
-      const arg_type<Timestamp>* /*timestamp1*/,
-      const arg_type<Timestamp>* /*timestamp2*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<Timestamp>* /*timestamp1*/,
+      const accessor_arg_type<Timestamp>* /*timestamp2*/) {
     if (unitString != nullptr) {
       unit_ = fromDateTimeUnitString(*unitString, false /*throwIfInvalid*/);
     }
@@ -1308,9 +1308,9 @@ struct DateDiffFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* unitString,
-      const arg_type<Date>* /*date1*/,
-      const arg_type<Date>* /*date2*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<Date>* /*date1*/,
+      const accessor_arg_type<Date>* /*date2*/) {
     if (unitString != nullptr) {
       unit_ = getDateUnit(*unitString, false);
     }
@@ -1319,9 +1319,9 @@ struct DateDiffFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* unitString,
-      const arg_type<TimestampWithTimezone>* /*timestampWithTimezone1*/,
-      const arg_type<TimestampWithTimezone>* /*timestampWithTimezone2*/) {
+      const accessor_arg_type<Varchar>* unitString,
+      const accessor_arg_type<TimestampWithTimezone>* /*timestampWithTimezone1*/,
+      const accessor_arg_type<TimestampWithTimezone>* /*timestampWithTimezone2*/) {
     if (unitString != nullptr) {
       unit_ = fromDateTimeUnitString(*unitString, false /*throwIfInvalid*/);
     }
@@ -1390,8 +1390,8 @@ struct DateFormatFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Timestamp>* /*timestamp*/,
-      const arg_type<Varchar>* formatString) {
+      const accessor_arg_type<Timestamp>* /*timestamp*/,
+      const accessor_arg_type<Varchar>* formatString) {
     sessionTimeZone_ = getTimeZoneFromConfig(config);
     if (formatString != nullptr) {
       setFormatter(*formatString);
@@ -1402,8 +1402,8 @@ struct DateFormatFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<TimestampWithTimezone>* /*timestamp*/,
-      const arg_type<Varchar>* formatString) {
+      const accessor_arg_type<TimestampWithTimezone>* /*timestamp*/,
+      const accessor_arg_type<Varchar>* formatString) {
     if (formatString != nullptr) {
       setFormatter(*formatString);
       isConstFormat_ = true;
@@ -1469,7 +1469,7 @@ struct FromIso8601Timestamp {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* /*input*/) {
+      const accessor_arg_type<Varchar>* /*input*/) {
     auto sessionTzName = config.sessionTimezone();
     if (!sessionTzName.empty()) {
       sessionTzID_ = tz::getTimeZoneID(sessionTzName);
@@ -1511,8 +1511,8 @@ struct DateParseFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* /*input*/,
-      const arg_type<Varchar>* formatString) {
+      const accessor_arg_type<Varchar>* /*input*/,
+      const accessor_arg_type<Varchar>* formatString) {
     if (formatString != nullptr) {
       format_ = buildMysqlDateTimeFormatter(
           std::string_view(formatString->data(), formatString->size()));
@@ -1555,8 +1555,8 @@ struct FormatDateTimeFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Timestamp>* /*timestamp*/,
-      const arg_type<Varchar>* formatString) {
+      const accessor_arg_type<Timestamp>* /*timestamp*/,
+      const accessor_arg_type<Varchar>* formatString) {
     sessionTimeZone_ = getTimeZoneFromConfig(config);
     if (formatString != nullptr) {
       setFormatter(*formatString);
@@ -1629,8 +1629,8 @@ struct ParseDateTimeFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* /*input*/,
-      const arg_type<Varchar>* format) {
+      const accessor_arg_type<Varchar>* /*input*/,
+      const accessor_arg_type<Varchar>* format) {
     if (format != nullptr) {
       format_ = buildJodaDateTimeFormatter(
           std::string_view(format->data(), format->size()));
@@ -1726,7 +1726,7 @@ struct ToISO8601Function {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& config,
-      const arg_type<Timestamp>* /*input*/) {
+      const accessor_arg_type<Timestamp>* /*input*/) {
     if (inputTypes[0]->isTimestamp()) {
       timeZone_ = getTimeZoneFromConfig(config);
     }
@@ -1780,8 +1780,8 @@ struct AtTimezoneFunction : public TimestampWithTimezoneSupport<T> {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<TimestampWithTimezone>* /*tsWithTz*/,
-      const arg_type<Varchar>* timezone) {
+      const accessor_arg_type<TimestampWithTimezone>* /*tsWithTz*/,
+      const accessor_arg_type<Varchar>* timezone) {
     if (timezone != nullptr) {
       targetTimezoneID_ = tz::getTimeZoneID(
           std::string_view(timezone->data(), timezone->size()));
diff --git a/velox/functions/prestosql/DecimalFunctions.cpp b/velox/functions/prestosql/DecimalFunctions.cpp
index 283d5e7ad9f68..c33b17150c756 100644
--- a/velox/functions/prestosql/DecimalFunctions.cpp
+++ b/velox/functions/prestosql/DecimalFunctions.cpp
@@ -32,8 +32,8 @@ struct DecimalPlusFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      A* /*a*/,
-      B* /*b*/) {
+      accessor_arg_type<A>* /*a*/,
+      accessor_arg_type<B>* /*b*/) {
     auto aType = inputTypes[0];
     auto bType = inputTypes[1];
     auto aScale = getDecimalPrecisionScale(*aType).second;
@@ -81,8 +81,8 @@ struct DecimalMinusFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      A* /*a*/,
-      B* /*b*/) {
+      accessor_arg_type<A>* /*a*/,
+      accessor_arg_type<B>* /*b*/) {
     const auto& aType = inputTypes[0];
     const auto& bType = inputTypes[1];
     auto aScale = getDecimalPrecisionScale(*aType).second;
@@ -141,8 +141,8 @@ struct DecimalDivideFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      A* /*a*/,
-      B* /*b*/) {
+      accessor_arg_type<A>* /*a*/,
+      accessor_arg_type<B>* /*b*/) {
     auto aType = inputTypes[0];
     auto bType = inputTypes[1];
     auto aScale = getDecimalPrecisionScale(*aType).second;
@@ -171,8 +171,8 @@ struct DecimalModulusFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      A* /*a*/,
-      B* /*b*/) {
+      accessor_arg_type<A>* /*a*/,
+      accessor_arg_type<B>* /*b*/) {
     const auto& aType = inputTypes[0];
     const auto& bType = inputTypes[1];
     auto [aPrecision, aScale] = getDecimalPrecisionScale(*aType);
@@ -221,7 +221,7 @@ struct DecimalRoundFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& config,
-      A* a) {
+      accessor_arg_type<A>* a) {
     initialize(inputTypes, config, a, nullptr);
   }
 
@@ -229,8 +229,8 @@ struct DecimalRoundFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      A* /*a*/,
-      const int32_t* /*n*/) {
+      accessor_arg_type<A>* /*a*/,
+      const accessor_arg_type<int32_t>* /*n*/) {
     const auto [precision, scale] = getDecimalPrecisionScale(*inputTypes[0]);
     precision_ = precision;
     scale_ = scale;
@@ -271,7 +271,7 @@ struct DecimalFloorFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      A* /*a*/) {
+      accessor_arg_type<A>* /*a*/) {
     scale_ = getDecimalPrecisionScale(*inputTypes[0]).second;
   }
 
@@ -296,7 +296,7 @@ struct DecimalTruncateFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& /*config*/,
-      A* /*a*/) {
+      accessor_arg_type<A>* /*a*/) {
     const auto [precision, scale] = getDecimalPrecisionScale(*inputTypes[0]);
     precision_ = precision;
     scale_ = scale;
@@ -306,8 +306,8 @@ struct DecimalTruncateFunction {
   void initialize(
       const std::vector<TypePtr>& inputTypes,
       const core::QueryConfig& config,
-      A* a,
-      const int32_t* /*n*/) {
+      accessor_arg_type<A>* a,
+      const accessor_arg_type<int32_t>* /*n*/) {
     initialize(inputTypes, config, a);
   }
 
diff --git a/velox/functions/prestosql/HyperLogLogFunctions.h b/velox/functions/prestosql/HyperLogLogFunctions.h
index 8dc85d1ef58fe..b46e4f81ab152 100644
--- a/velox/functions/prestosql/HyperLogLogFunctions.h
+++ b/velox/functions/prestosql/HyperLogLogFunctions.h
@@ -64,7 +64,7 @@ struct EmptyApproxSetWithMaxErrorFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const double* maxStandardError) {
+      const accessor_arg_type<double>* maxStandardError) {
     VELOX_USER_CHECK_NOT_NULL(
         maxStandardError,
         "empty_approx_set function requires constant value for maxStandardError argument");
diff --git a/velox/functions/prestosql/MapSubset.h b/velox/functions/prestosql/MapSubset.h
index 238922e8f0529..21560cf3dbbda 100644
--- a/velox/functions/prestosql/MapSubset.h
+++ b/velox/functions/prestosql/MapSubset.h
@@ -29,8 +29,8 @@ struct MapSubsetPrimitiveFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Map<Key, Generic<T1>>>* /*inputMap*/,
-      const arg_type<Array<Key>>* keys) {
+      const accessor_arg_type<Map<Key, Generic<T1>>>* /*inputMap*/,
+      const accessor_arg_type<Array<Key>>* keys) {
     if (keys != nullptr) {
       constantSearchKeys_ = true;
       initializeSearchKeys(*keys);
@@ -96,8 +96,8 @@ struct MapSubsetVarcharFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Map<Varchar, Generic<T1>>>* /*inputMap*/,
-      const arg_type<Array<Varchar>>* keys) {
+      const accessor_arg_type<Map<Varchar, Generic<T1>>>* /*inputMap*/,
+      const accessor_arg_type<Array<Varchar>>* keys) {
     if (keys != nullptr) {
       constantSearchKeys_ = true;
 
diff --git a/velox/functions/prestosql/StringFunctions.h b/velox/functions/prestosql/StringFunctions.h
index 22d3ea26ccb8c..b3bb21ef9c5f2 100644
--- a/velox/functions/prestosql/StringFunctions.h
+++ b/velox/functions/prestosql/StringFunctions.h
@@ -545,8 +545,8 @@ struct NormalizeFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* /*string*/,
-      const arg_type<Varchar>* form) {
+      const accessor_arg_type<Varchar>* /*string*/,
+      const accessor_arg_type<Varchar>* form) {
     VELOX_USER_CHECK_NOT_NULL(form);
     VELOX_USER_CHECK_NE(
         normalizationOptions.count(*form),
diff --git a/velox/functions/sparksql/DateTimeFunctions.h b/velox/functions/sparksql/DateTimeFunctions.h
index 7b9652e586b9b..1e6df07b99dc5 100644
--- a/velox/functions/sparksql/DateTimeFunctions.h
+++ b/velox/functions/sparksql/DateTimeFunctions.h
@@ -155,7 +155,7 @@ struct UnixTimestampParseFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* /*input*/) {
+      const accessor_arg_type<Varchar>* /*input*/) {
     format_ = buildJodaDateTimeFormatter(kDefaultFormat_);
     setTimezone(config);
   }
@@ -205,8 +205,8 @@ struct UnixTimestampParseWithFormatFunction
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* /*input*/,
-      const arg_type<Varchar>* format) {
+      const accessor_arg_type<Varchar>* /*input*/,
+      const accessor_arg_type<Varchar>* format) {
     if (format != nullptr) {
       try {
         this->format_ = buildJodaDateTimeFormatter(
@@ -260,8 +260,8 @@ struct FromUnixtimeFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<int64_t>* /*unixtime*/,
-      const arg_type<Varchar>* format) {
+      const accessor_arg_type<int64_t>* /*unixtime*/,
+      const accessor_arg_type<Varchar>* format) {
     sessionTimeZone_ = getTimeZoneFromConfig(config);
     if (format != nullptr) {
       setFormatter(*format);
@@ -304,8 +304,8 @@ struct ToUtcTimestampFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* /*input*/,
-      const arg_type<Varchar>* timezone) {
+      const accessor_arg_type<Varchar>* /*input*/,
+      const accessor_arg_type<Varchar>* timezone) {
     if (timezone) {
       tzID_ = tz::getTimeZoneID(
           std::string_view((*timezone).data(), (*timezone).size()), false);
@@ -337,8 +337,8 @@ struct FromUtcTimestampFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* /*input*/,
-      const arg_type<Varchar>* timezone) {
+      const accessor_arg_type<Varchar>* /*input*/,
+      const accessor_arg_type<Varchar>* timezone) {
     if (timezone) {
       tzID_ = tz::getTimeZoneID(
           std::string_view((*timezone).data(), (*timezone).size()), false);
@@ -370,8 +370,8 @@ struct GetTimestampFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const arg_type<Varchar>* /*input*/,
-      const arg_type<Varchar>* format) {
+      const accessor_arg_type<Varchar>* /*input*/,
+      const accessor_arg_type<Varchar>* format) {
     auto sessionTimezoneName = config.sessionTimezone();
     if (!sessionTimezoneName.empty()) {
       sessionTimezoneId_ = tz::getTimeZoneID(sessionTimezoneName);
@@ -650,8 +650,8 @@ struct NextDayFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Date>* /*startDate*/,
-      const arg_type<Varchar>* dayOfWeek) {
+      const accessor_arg_type<Date>* /*startDate*/,
+      const accessor_arg_type<Varchar>* dayOfWeek) {
     if (dayOfWeek != nullptr) {
       weekDay_ = getDayOfWeekFromString(*dayOfWeek);
       if (!weekDay_.has_value()) {
diff --git a/velox/functions/sparksql/In.cpp b/velox/functions/sparksql/In.cpp
index cddfbaab9ed7a..29b7d9cbe949e 100644
--- a/velox/functions/sparksql/In.cpp
+++ b/velox/functions/sparksql/In.cpp
@@ -83,8 +83,8 @@ struct InFunctionOuter {
     FOLLY_ALWAYS_INLINE void initialize(
         const std::vector<TypePtr>& /*inputTypes*/,
         const core::QueryConfig& /*config*/,
-        const arg_type<TInput>* /*searchTerm*/,
-        const arg_type<velox::Array<TInput>>* searchElements) {
+        const accessor_arg_type<TInput>* /*searchTerm*/,
+        const accessor_arg_type<velox::Array<TInput>>* searchElements) {
       if (searchElements == nullptr) {
         return;
       }
diff --git a/velox/functions/sparksql/MightContain.h b/velox/functions/sparksql/MightContain.h
index 21b16007b8bdb..036837ebc6103 100644
--- a/velox/functions/sparksql/MightContain.h
+++ b/velox/functions/sparksql/MightContain.h
@@ -28,8 +28,8 @@ struct BloomFilterMightContainFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig&,
-      const arg_type<Varbinary>* serialized,
-      const arg_type<int64_t>*) {
+      const accessor_arg_type<Varbinary>* serialized,
+      const accessor_arg_type<int64_t>*) {
     if (serialized != nullptr) {
       bloomFilter_.merge(serialized->str().c_str());
     }
diff --git a/velox/functions/sparksql/Rand.h b/velox/functions/sparksql/Rand.h
index 01cf174861a83..b782b34589e43 100644
--- a/velox/functions/sparksql/Rand.h
+++ b/velox/functions/sparksql/Rand.h
@@ -21,13 +21,15 @@ namespace facebook::velox::functions::sparksql {
 
 template <typename T>
 struct RandFunction {
+  VELOX_DEFINE_FUNCTION_TYPES(T);
+
   static constexpr bool is_deterministic = false;
 
   template <typename TInput>
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const TInput* seedInput) {
+      const accessor_arg_type<TInput>* seedInput) {
     const auto partitionId = config.sparkPartitionId();
     int64_t seed = seedInput ? (int64_t)*seedInput : 0;
     generator_.seed(seed + partitionId);
diff --git a/velox/functions/sparksql/Size.cpp b/velox/functions/sparksql/Size.cpp
index 70d7d3d509d2d..f363fd9a0211a 100644
--- a/velox/functions/sparksql/Size.cpp
+++ b/velox/functions/sparksql/Size.cpp
@@ -32,7 +32,7 @@ struct Size {
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
       const TInput* /*input*/,
-      const bool* legacySizeOfNull) {
+      const accessor_arg_type<bool>* legacySizeOfNull) {
     if (legacySizeOfNull == nullptr) {
       VELOX_USER_FAIL("Constant legacySizeOfNull is expected.");
     }
diff --git a/velox/functions/sparksql/String.h b/velox/functions/sparksql/String.h
index 2680c36b75571..fce7fd9ef0fb3 100644
--- a/velox/functions/sparksql/String.h
+++ b/velox/functions/sparksql/String.h
@@ -832,9 +832,9 @@ struct TranslateFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& /*config*/,
-      const arg_type<Varchar>* /*string*/,
-      const arg_type<Varchar>* match,
-      const arg_type<Varchar>* replace) {
+      const accessor_arg_type<Varchar>* /*string*/,
+      const accessor_arg_type<Varchar>* match,
+      const accessor_arg_type<Varchar>* replace) {
     if (match != nullptr && replace != nullptr) {
       isConstantDictionary_ = true;
     }
diff --git a/velox/functions/sparksql/Uuid.h b/velox/functions/sparksql/Uuid.h
index 1e403decb1d3d..2b87e0910dbca 100644
--- a/velox/functions/sparksql/Uuid.h
+++ b/velox/functions/sparksql/Uuid.h
@@ -34,7 +34,7 @@ struct UuidFunction {
   void initialize(
       const std::vector<TypePtr>& /*inputTypes*/,
       const core::QueryConfig& config,
-      const int64_t* seed) {
+      const accessor_arg_type<int64_t>* seed) {
     VELOX_CHECK_NOT_NULL(seed, "seed argument must be constant");
     const int32_t partitionId = config.sparkPartitionId();
     generator_.seed((*seed) + partitionId);