diff --git a/velox/core/SimpleFunctionMetadata.h b/velox/core/SimpleFunctionMetadata.h
index 0d9cc0f8b4bc..9662ef9e9086 100644
--- a/velox/core/SimpleFunctionMetadata.h
+++ b/velox/core/SimpleFunctionMetadata.h
@@ -806,7 +806,7 @@ class UDFHolder {
       void,
       const std::vector<TypePtr>&,
       const core::QueryConfig&,
-      const exec_arg_type<TArgs>*...>::value;
+      const ConstantArg<exec_arg_type<TArgs>*...>::value>*;
 
   // TODO Remove
   static constexpr bool udf_has_legacy_initialize = util::has_method<
diff --git a/velox/functions/lib/Re2Functions.h b/velox/functions/lib/Re2Functions.h
index 5e7267c7ea83..ea178a2ed392 100644
--- a/velox/functions/lib/Re2Functions.h
+++ b/velox/functions/lib/Re2Functions.h
@@ -277,11 +277,11 @@ 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 ConstantArg<arg_type<Varchar>>* /*string*/,
+      const ConstantArg<arg_type<Varchar>>* pattern,
+      const ConstantArg<arg_type<Varchar>>* replacement) {
     if (pattern != nullptr) {
-      const auto processedPattern = prepareRegexpPattern(*pattern);
+      const auto processedPattern = prepareRegexpPattern(*pattern->value());
       re_.emplace(processedPattern, RE2::Quiet);
       VELOX_USER_CHECK(
           re_->ok(),
@@ -294,7 +294,7 @@ struct Re2RegexpReplace {
       // Constant 'replacement' with non-constant 'pattern' needs to be
       // processed separately for each row.
       if (pattern != nullptr) {
-        ensureProcessedReplacement(re_.value(), *replacement);
+        ensureProcessedReplacement(re_.value(), *replacement->value());
         constantReplacement_ = true;
       }
     }
@@ -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 ConstantArg<arg_type<Varchar>>* string,
+      const ConstantArg<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 39892124e0a7..e5b8aba45e60 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 ConstantArg<arg_type<Timestamp>>* /*timestamp*/) {
     timeZone_ = getTimeZoneFromConfig(config);
   }
 };
diff --git a/velox/functions/prestosql/DateTimeFunctions.h b/velox/functions/prestosql/DateTimeFunctions.h
index 51eb5f03d89f..d4a2edfee10a 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 ConstantArg<arg_type<double>>* /*unixtime*/,
+      const ConstantArg<arg_type<Varchar>>* timezone) {
     if (timezone != nullptr) {
       tzID_ = tz::getTimeZoneID((std::string_view)(*timezone));
     }
@@ -79,9 +79,9 @@ struct FromUnixtimeFunction {
   FOLLY_ALWAYS_INLINE void initialize(
       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 ConstantArg<arg_type<double>>* /*unixtime*/,
+      const ConstantArg<arg_type<int64_t>>* hours,
+      const ConstantArg<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 ConstantArg<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 ConstantArg<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 ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<int64_t>* /*value*/,
+      const ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<int64_t>* /*value*/,
+      const ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<arg_type<Timestamp>>* /*timestamp1*/,
+      const ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<arg_type<Date>>* /*date1*/,
+      const ConstantArg<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 ConstantArg<arg_type<Varchar>>* unitString,
+      const ConstantArg<arg_type<TimestampWithTimezone>>* /*timestampWithTimezone1*/,
+      const ConstantArg<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 ConstantArg<arg_type<Timestamp>>* /*timestamp*/,
+      const ConstantArg<arg_type<Varchar>>* formatString) {
     sessionTimeZone_ = getTimeZoneFromConfig(config);
     if (formatString != nullptr) {
       setFormatter(*formatString);