diff --git a/velox/functions/prestosql/aggregates/ValueList.cpp b/velox/functions/prestosql/aggregates/ValueList.cpp
index 89da682020080..4c2433809a3f2 100644
--- a/velox/functions/prestosql/aggregates/ValueList.cpp
+++ b/velox/functions/prestosql/aggregates/ValueList.cpp
@@ -123,4 +123,27 @@ bool ValueListReader::next(BaseVector& output, vector_size_t outputIndex) {
   pos_++;
   return pos_ < size_;
 }
+
+bool ValueListReader::nextIgnoreNull(
+    BaseVector& output,
+    vector_size_t outputIndex,
+    bool& skipped) {
+  if (pos_ == lastNullsStart_) {
+    nulls_ = lastNulls_;
+  } else if (pos_ % 64 == 0) {
+    nulls_ = nullsStream_.read<uint64_t>();
+  }
+
+  if (nulls_ & (1UL << (pos_ % 64))) {
+    // Ignore null.
+    skipped = true;
+  } else {
+    exec::ContainerRowSerde::instance().deserialize(
+        dataStream_, outputIndex, &output);
+  }
+
+  pos_++;
+  return pos_ < size_;
+}
+
 } // namespace facebook::velox::aggregate
diff --git a/velox/functions/prestosql/aggregates/ValueList.h b/velox/functions/prestosql/aggregates/ValueList.h
index 13271b2856927..9eb5d17e008b9 100644
--- a/velox/functions/prestosql/aggregates/ValueList.h
+++ b/velox/functions/prestosql/aggregates/ValueList.h
@@ -108,6 +108,8 @@ class ValueListReader {
   explicit ValueListReader(ValueList& values);
 
   bool next(BaseVector& output, vector_size_t outputIndex);
+  bool
+  nextIgnoreNull(BaseVector& output, vector_size_t outputIndex, bool& skipped);
 
  private:
   const vector_size_t size_;
diff --git a/velox/functions/sparksql/aggregates/ArrayAggAggregate.cpp b/velox/functions/sparksql/aggregates/ArrayAggAggregate.cpp
new file mode 100644
index 0000000000000..651a72c122adb
--- /dev/null
+++ b/velox/functions/sparksql/aggregates/ArrayAggAggregate.cpp
@@ -0,0 +1,211 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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.
+ */
+#include "velox/exec/ContainerRowSerde.h"
+#include "velox/expression/FunctionSignature.h"
+#include "velox/functions/prestosql/aggregates/AggregateNames.h"
+#include "velox/functions/prestosql/aggregates/ValueList.h"
+#include "velox/vector/ComplexVector.h"
+
+namespace facebook::velox::functions::aggregate::sparksql {
+namespace {
+
+using namespace facebook::velox::aggregate;
+
+struct ArrayAccumulator {
+  ValueList elements;
+};
+
+class ArrayAggAggregate : public exec::Aggregate {
+ public:
+  explicit ArrayAggAggregate(TypePtr resultType) : Aggregate(resultType) {}
+
+  int32_t accumulatorFixedWidthSize() const override {
+    return sizeof(ArrayAccumulator);
+  }
+
+  bool isFixedSize() const override {
+    return false;
+  }
+
+  void initializeNewGroups(
+      char** groups,
+      folly::Range<const vector_size_t*> indices) override {
+    for (auto index : indices) {
+      new (groups[index] + offset_) ArrayAccumulator();
+    }
+  }
+
+  void extractValues(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    auto vector = (*result)->as<ArrayVector>();
+    VELOX_CHECK(vector);
+    vector->resize(numGroups);
+
+    auto elements = vector->elements();
+    elements->resize(countElements(groups, numGroups));
+
+    uint64_t* rawNulls = getRawNulls(vector);
+    vector_size_t offset = 0;
+    for (int32_t i = 0; i < numGroups; ++i) {
+      // No null result, either empty array or non-empty array.
+      clearNull(rawNulls, i);
+      auto& values = value<ArrayAccumulator>(groups[i])->elements;
+      auto arraySize = values.size();
+      if (arraySize) {
+        ValueListReader reader(values);
+        int count = 0;
+        for (auto index = 0; index < arraySize; ++index) {
+          // To mark whether skipped due to null input value.
+          bool skipped = false;
+          reader.nextIgnoreNull(*elements, offset + count, skipped);
+          if (!skipped) {
+            count++;
+          }
+        }
+        vector->setOffsetAndSize(i, offset, count);
+        offset += count;
+      } else {
+        // Empty array.
+        vector->setOffsetAndSize(i, offset, 0);
+      }
+    }
+  }
+
+  void extractAccumulators(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    extractValues(groups, numGroups, result);
+  }
+
+  void addRawInput(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    decodedElements_.decode(*args[0], rows);
+    rows.applyToSelected([&](vector_size_t row) {
+      auto group = groups[row];
+      auto tracker = trackRowSize(group);
+      value<ArrayAccumulator>(group)->elements.appendValue(
+          decodedElements_, row, allocator_);
+    });
+  }
+
+  void addIntermediateResults(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    decodedIntermediate_.decode(*args[0], rows);
+
+    auto arrayVector = decodedIntermediate_.base()->as<ArrayVector>();
+    auto& elements = arrayVector->elements();
+    rows.applyToSelected([&](vector_size_t row) {
+      auto group = groups[row];
+      auto decodedRow = decodedIntermediate_.index(row);
+      auto tracker = trackRowSize(group);
+      value<ArrayAccumulator>(group)->elements.appendRange(
+          elements,
+          arrayVector->offsetAt(decodedRow),
+          arrayVector->sizeAt(decodedRow),
+          allocator_);
+    });
+  }
+
+  void addSingleGroupRawInput(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /* mayPushdown */) override {
+    auto& values = value<ArrayAccumulator>(group)->elements;
+
+    decodedElements_.decode(*args[0], rows);
+    auto tracker = trackRowSize(group);
+    rows.applyToSelected([&](vector_size_t row) {
+      values.appendValue(decodedElements_, row, allocator_);
+    });
+  }
+
+  void addSingleGroupIntermediateResults(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /* mayPushdown */) override {
+    decodedIntermediate_.decode(*args[0], rows);
+    auto arrayVector = decodedIntermediate_.base()->as<ArrayVector>();
+
+    auto& values = value<ArrayAccumulator>(group)->elements;
+    auto elements = arrayVector->elements();
+    rows.applyToSelected([&](vector_size_t row) {
+      auto decodedRow = decodedIntermediate_.index(row);
+      values.appendRange(
+          elements,
+          arrayVector->offsetAt(decodedRow),
+          arrayVector->sizeAt(decodedRow),
+          allocator_);
+    });
+  }
+
+  void destroy(folly::Range<char**> groups) override {
+    for (auto group : groups) {
+      value<ArrayAccumulator>(group)->elements.free(allocator_);
+    }
+  }
+
+ private:
+  vector_size_t countElements(char** groups, int32_t numGroups) const {
+    vector_size_t size = 0;
+    for (int32_t i = 0; i < numGroups; ++i) {
+      size += value<ArrayAccumulator>(groups[i])->elements.size();
+    }
+    return size;
+  }
+
+  // Reusable instance of DecodedVector for decoding input vectors.
+  DecodedVector decodedElements_;
+  DecodedVector decodedIntermediate_;
+};
+
+exec::AggregateRegistrationResult registerArray(const std::string& name) {
+  std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures{
+      exec::AggregateFunctionSignatureBuilder()
+          .typeVariable("E")
+          .returnType("array(E)")
+          .intermediateType("array(E)")
+          .argumentType("E")
+          .build()};
+
+  return exec::registerAggregateFunction(
+      name,
+      std::move(signatures),
+      [name](
+          core::AggregationNode::Step step,
+          const std::vector<TypePtr>& argTypes,
+          const TypePtr& resultType) -> std::unique_ptr<exec::Aggregate> {
+        VELOX_CHECK_EQ(
+            argTypes.size(), 1, "{} takes at most one argument", name);
+        return std::make_unique<ArrayAggAggregate>(resultType);
+      },
+      false,
+      true);
+}
+
+} // namespace
+
+void registerArrayAggregate(const std::string& prefix) {
+  registerArray(prefix + kArrayAgg);
+}
+
+} // namespace facebook::velox::functions::aggregate::sparksql
diff --git a/velox/functions/sparksql/aggregates/CMakeLists.txt b/velox/functions/sparksql/aggregates/CMakeLists.txt
index 4977cea9ea7c7..03da0b966d6b7 100644
--- a/velox/functions/sparksql/aggregates/CMakeLists.txt
+++ b/velox/functions/sparksql/aggregates/CMakeLists.txt
@@ -13,8 +13,12 @@
 # limitations under the License.
 add_library(
   velox_functions_spark_aggregates
-  BitwiseXorAggregate.cpp BloomFilterAggAggregate.cpp FirstLastAggregate.cpp
-  Register.cpp)
+  ArrayAggAggregate.cpp
+  BitwiseXorAggregate.cpp
+  BloomFilterAggAggregate.cpp
+  FirstLastAggregate.cpp
+  Register.cpp
+  SetAggAggregate.cpp)
 
 target_link_libraries(velox_functions_spark_aggregates fmt::fmt velox_exec
                       velox_expression_functions velox_aggregates velox_vector)
diff --git a/velox/functions/sparksql/aggregates/Register.cpp b/velox/functions/sparksql/aggregates/Register.cpp
index f1983111c028a..176a6feead6b1 100644
--- a/velox/functions/sparksql/aggregates/Register.cpp
+++ b/velox/functions/sparksql/aggregates/Register.cpp
@@ -25,12 +25,16 @@ namespace facebook::velox::functions::aggregate::sparksql {
 using namespace facebook::velox::functions::sparksql::aggregates;
 
 extern void registerFirstLastAggregates(const std::string& prefix);
+extern void registerArrayAggregate(const std::string& prefix);
+extern void registerSetAggAggregate(const std::string& prefix);
 
 void registerAggregateFunctions(const std::string& prefix) {
+  registerArrayAggregate(prefix);
   registerFirstLastAggregates(prefix);
   registerBitwiseXorAggregate(prefix);
   registerBloomFilterAggAggregate(prefix + "bloom_filter_agg");
   registerDecimalAvgAggregate(prefix + "decimal_avg");
   registerDecimalSumAggregate(prefix + "decimal_sum");
+  registerSetAggAggregate(prefix);
 }
 } // namespace facebook::velox::functions::aggregate::sparksql
diff --git a/velox/functions/sparksql/aggregates/SetAggAggregate.cpp b/velox/functions/sparksql/aggregates/SetAggAggregate.cpp
new file mode 100644
index 0000000000000..9052e5ebc91f9
--- /dev/null
+++ b/velox/functions/sparksql/aggregates/SetAggAggregate.cpp
@@ -0,0 +1,451 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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.
+ */
+#include "velox/exec/Aggregate.h"
+#include "velox/functions/prestosql/aggregates/AggregateNames.h"
+#include "velox/functions/prestosql/aggregates/Strings.h"
+#include "velox/vector/FlatVector.h"
+
+namespace facebook::velox::functions::aggregate::sparksql {
+
+namespace {
+
+using namespace facebook::velox::aggregate;
+using namespace facebook::velox::aggregate::prestosql;
+
+/// Maintains a set of unique values of fixed-width type (integers). Also
+/// maintains a flag indicating whether there was a null value.
+template <typename T>
+struct Accumulator {
+  // bool hasNull{false};
+  folly::
+      F14FastSet<T, std::hash<T>, std::equal_to<T>, AlignedStlAllocator<T, 16>>
+          uniqueValues;
+
+  explicit Accumulator(HashStringAllocator* allocator)
+      : uniqueValues{AlignedStlAllocator<T, 16>(allocator)} {}
+
+  /// Adds value if new. No-op if the value was added before.
+  void addValue(
+      const DecodedVector& decoded,
+      vector_size_t index,
+      HashStringAllocator* /*allocator*/) {
+    if (decoded.isNullAt(index)) {
+      // Ignore null.
+      // hasNull = true;
+    } else {
+      uniqueValues.insert(decoded.valueAt<T>(index));
+    }
+  }
+
+  /// Adds new values from an array.
+  void addValues(
+      const ArrayVector& arrayVector,
+      vector_size_t index,
+      const DecodedVector& values,
+      HashStringAllocator* allocator) {
+    const auto size = arrayVector.sizeAt(index);
+    const auto offset = arrayVector.offsetAt(index);
+
+    for (auto i = 0; i < size; ++i) {
+      addValue(values, offset + i, allocator);
+    }
+  }
+
+  /// Returns number of unique values including null.
+  size_t size() const {
+    // return uniqueValues.size() + (hasNull ? 1 : 0);
+    return uniqueValues.size();
+  }
+
+  /// Copies the unique values and null into the specified vector starting at
+  /// the specified offset.
+  vector_size_t extractValues(FlatVector<T>& values, vector_size_t offset) {
+    vector_size_t index = offset;
+    for (auto value : uniqueValues) {
+      values.set(index++, value);
+    }
+
+    // if (hasNull) {
+    //   values.setNull(index++, true);
+    // }
+
+    return index - offset;
+  }
+};
+
+/// Maintains a set of unique strings.
+struct StringViewAccumulator {
+  /// A set of unique StringViews pointing to storage managed by 'strings'.
+  Accumulator<StringView> base;
+
+  /// Stores unique non-null non-inline strings.
+  Strings strings;
+
+  explicit StringViewAccumulator(HashStringAllocator* allocator)
+      : base{allocator} {}
+
+  void addValue(
+      const DecodedVector& decoded,
+      vector_size_t index,
+      HashStringAllocator* allocator) {
+    if (decoded.isNullAt(index)) {
+      // Ignore null.
+      // base.hasNull = true;
+    } else {
+      auto value = decoded.valueAt<StringView>(index);
+      if (!value.isInline()) {
+        if (base.uniqueValues.contains(value)) {
+          return;
+        }
+        value = strings.append(value, *allocator);
+      }
+      base.uniqueValues.insert(value);
+    }
+  }
+
+  void addValues(
+      const ArrayVector& arrayVector,
+      vector_size_t index,
+      const DecodedVector& values,
+      HashStringAllocator* allocator) {
+    const auto size = arrayVector.sizeAt(index);
+    const auto offset = arrayVector.offsetAt(index);
+
+    for (auto i = 0; i < size; ++i) {
+      addValue(values, offset + i, allocator);
+    }
+  }
+
+  size_t size() const {
+    return base.size();
+  }
+
+  vector_size_t extractValues(
+      FlatVector<StringView>& values,
+      vector_size_t offset) {
+    return base.extractValues(values, offset);
+  }
+};
+
+template <typename T>
+struct AccumulatorTypeTraits {
+  using AccumulatorType = Accumulator<T>;
+};
+
+template <>
+struct AccumulatorTypeTraits<StringView> {
+  using AccumulatorType = StringViewAccumulator;
+};
+
+template <typename T>
+class SetBaseAggregate : public exec::Aggregate {
+ public:
+  explicit SetBaseAggregate(const TypePtr& resultType)
+      : exec::Aggregate(resultType) {}
+
+  using AccumulatorType = typename AccumulatorTypeTraits<T>::AccumulatorType;
+
+  int32_t accumulatorFixedWidthSize() const override {
+    return sizeof(AccumulatorType);
+  }
+
+  bool isFixedSize() const override {
+    return false;
+  }
+
+  void initializeNewGroups(
+      char** groups,
+      folly::Range<const vector_size_t*> indices) override {
+    exec::Aggregate::setAllNulls(groups, indices);
+    for (auto i : indices) {
+      new (groups[i] + offset_) AccumulatorType(allocator_);
+    }
+  }
+
+  void extractValues(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    auto arrayVector = (*result)->as<ArrayVector>();
+    arrayVector->resize(numGroups);
+
+    auto* rawOffsets = arrayVector->offsets()->asMutable<vector_size_t>();
+    auto* rawSizes = arrayVector->sizes()->asMutable<vector_size_t>();
+
+    vector_size_t numValues = 0;
+    uint64_t* rawNulls = getRawNulls(arrayVector);
+    for (auto i = 0; i < numGroups; ++i) {
+      auto* group = groups[i];
+      if (isNull(group)) {
+        // arrayVector->setNull(i, true);
+      } else {
+        clearNull(rawNulls, i);
+
+        const auto size = value(group)->size();
+
+        rawOffsets[i] = numValues;
+        rawSizes[i] = size;
+
+        numValues += size;
+      }
+    }
+
+    auto values = arrayVector->elements()->as<FlatVector<T>>();
+    values->resize(numValues);
+
+    vector_size_t offset = 0;
+    for (auto i = 0; i < numGroups; ++i) {
+      auto* group = groups[i];
+      if (!isNull(group)) {
+        offset += value(group)->extractValues(*values, offset);
+      }
+    }
+  }
+
+  void extractAccumulators(char** groups, int32_t numGroups, VectorPtr* result)
+      override {
+    return extractValues(groups, numGroups, result);
+  }
+
+  void addIntermediateResults(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    decoded_.decode(*args[0], rows);
+
+    auto baseArray = decoded_.base()->template as<ArrayVector>();
+    decodedElements_.decode(*baseArray->elements());
+
+    rows.applyToSelected([&](vector_size_t i) {
+      if (decoded_.isNullAt(i)) {
+        return;
+      }
+
+      auto* group = groups[i];
+      clearNull(group);
+
+      auto tracker = trackRowSize(group);
+
+      auto decodedIndex = decoded_.index(i);
+      value(group)->addValues(
+          *baseArray, decodedIndex, decodedElements_, allocator_);
+    });
+  }
+
+  void addSingleGroupIntermediateResults(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    decoded_.decode(*args[0], rows);
+
+    auto baseArray = decoded_.base()->template as<ArrayVector>();
+
+    decodedElements_.decode(*baseArray->elements());
+
+    auto* accumulator = value(group);
+
+    auto tracker = trackRowSize(group);
+    rows.applyToSelected([&](vector_size_t i) {
+      if (decoded_.isNullAt(i)) {
+        return;
+      }
+
+      clearNull(group);
+
+      auto decodedIndex = decoded_.index(i);
+      accumulator->addValues(
+          *baseArray, decodedIndex, decodedElements_, allocator_);
+    });
+  }
+
+  void destroy(folly::Range<char**> groups) override {
+    if constexpr (std::is_same_v<T, StringView>) {
+      for (auto* group : groups) {
+        if (!isNull(group)) {
+          value(group)->strings.free(*allocator_);
+        }
+      }
+    }
+  }
+
+ protected:
+  inline AccumulatorType* value(char* group) {
+    return reinterpret_cast<AccumulatorType*>(group + Aggregate::offset_);
+  }
+
+  DecodedVector decoded_;
+  DecodedVector decodedElements_;
+};
+
+template <typename T>
+class SetAggAggregate : public SetBaseAggregate<T> {
+ public:
+  explicit SetAggAggregate(const TypePtr& resultType)
+      : SetBaseAggregate<T>(resultType) {}
+
+  using Base = SetBaseAggregate<T>;
+
+  void addRawInput(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    Base::decoded_.decode(*args[0], rows);
+
+    rows.applyToSelected([&](vector_size_t i) {
+      auto* group = groups[i];
+      Base::clearNull(group);
+
+      auto tracker = Base::trackRowSize(group);
+      Base::value(group)->addValue(Base::decoded_, i, Base::allocator_);
+    });
+  }
+
+  void addSingleGroupRawInput(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool /*mayPushdown*/) override {
+    Base::decoded_.decode(*args[0], rows);
+
+    Base::clearNull(group);
+    auto* accumulator = Base::value(group);
+
+    auto tracker = Base::trackRowSize(group);
+    rows.applyToSelected([&](vector_size_t i) {
+      accumulator->addValue(Base::decoded_, i, Base::allocator_);
+    });
+  }
+};
+
+template <typename T>
+class SetUnionAggregate : public SetBaseAggregate<T> {
+ public:
+  explicit SetUnionAggregate(const TypePtr& resultType)
+      : SetBaseAggregate<T>(resultType) {}
+
+  using Base = SetBaseAggregate<T>;
+
+  void addRawInput(
+      char** groups,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    Base::addIntermediateResults(groups, rows, args, mayPushdown);
+  }
+
+  void addSingleGroupRawInput(
+      char* group,
+      const SelectivityVector& rows,
+      const std::vector<VectorPtr>& args,
+      bool mayPushdown) override {
+    Base::addSingleGroupIntermediateResults(group, rows, args, mayPushdown);
+  }
+};
+
+template <template <typename T> class Aggregate>
+std::unique_ptr<exec::Aggregate> create(
+    TypeKind typeKind,
+    const TypePtr& resultType) {
+  switch (typeKind) {
+    case TypeKind::TINYINT:
+      return std::make_unique<Aggregate<int8_t>>(resultType);
+    case TypeKind::SMALLINT:
+      return std::make_unique<Aggregate<int16_t>>(resultType);
+    case TypeKind::INTEGER:
+      return std::make_unique<Aggregate<int32_t>>(resultType);
+    case TypeKind::BIGINT:
+      return std::make_unique<Aggregate<int64_t>>(resultType);
+    case TypeKind::VARCHAR:
+      return std::make_unique<Aggregate<StringView>>(resultType);
+    default:
+      VELOX_UNREACHABLE();
+  }
+}
+
+std::vector<std::string> supportedTypes() {
+  return {"tinyint", "smallint", "integer", "bigint", "varchar"};
+}
+
+exec::AggregateRegistrationResult registerSetAgg(const std::string& name) {
+  std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures;
+  for (const auto& inputType : supportedTypes()) {
+    const std::string arrayType = fmt::format("array({})", inputType);
+    signatures.push_back(exec::AggregateFunctionSignatureBuilder()
+                             .returnType(arrayType)
+                             .intermediateType(arrayType)
+                             .argumentType(inputType)
+                             .build());
+  }
+
+  return exec::registerAggregateFunction(
+      name,
+      std::move(signatures),
+      [name](
+          core::AggregationNode::Step step,
+          const std::vector<TypePtr>& argTypes,
+          const TypePtr& resultType) -> std::unique_ptr<exec::Aggregate> {
+        VELOX_CHECK_EQ(argTypes.size(), 1);
+
+        const TypeKind typeKind = exec::isRawInput(step)
+            ? argTypes[0]->kind()
+            : argTypes[0]->childAt(0)->kind();
+
+        return create<SetAggAggregate>(typeKind, resultType);
+      },
+      false,
+      true);
+}
+
+exec::AggregateRegistrationResult registerSetUnion(const std::string& name) {
+  std::vector<std::shared_ptr<exec::AggregateFunctionSignature>> signatures;
+  for (const auto& type : supportedTypes()) {
+    const std::string arrayType = fmt::format("array({})", type);
+    signatures.push_back(exec::AggregateFunctionSignatureBuilder()
+                             .returnType(arrayType)
+                             .intermediateType(arrayType)
+                             .argumentType(arrayType)
+                             .build());
+  }
+
+  return exec::registerAggregateFunction(
+      name,
+      std::move(signatures),
+      [name](
+          core::AggregationNode::Step /*step*/,
+          const std::vector<TypePtr>& argTypes,
+          const TypePtr& resultType) -> std::unique_ptr<exec::Aggregate> {
+        VELOX_CHECK_EQ(argTypes.size(), 1);
+
+        const TypeKind typeKind = argTypes[0]->childAt(0)->kind();
+
+        return create<SetUnionAggregate>(typeKind, resultType);
+      },
+      false,
+      true);
+}
+
+} // namespace
+
+void registerSetAggAggregate(const std::string& prefix) {
+  registerSetAgg(prefix + kSetAgg);
+}
+
+void registerSetUnionAggregate(const std::string& prefix) {
+  registerSetUnion(prefix + kSetUnion);
+}
+
+} // namespace facebook::velox::functions::aggregate::sparksql
diff --git a/velox/functions/sparksql/aggregates/tests/ArrayAggTest.cpp b/velox/functions/sparksql/aggregates/tests/ArrayAggTest.cpp
new file mode 100644
index 0000000000000..8b222ea7ab327
--- /dev/null
+++ b/velox/functions/sparksql/aggregates/tests/ArrayAggTest.cpp
@@ -0,0 +1,179 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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.
+ */
+#include "velox/dwio/common/tests/utils/BatchMaker.h"
+#include "velox/exec/tests/utils/AssertQueryBuilder.h"
+#include "velox/exec/tests/utils/Cursor.h"
+#include "velox/exec/tests/utils/PlanBuilder.h"
+#include "velox/functions/lib/aggregates/tests/AggregationTestBase.h"
+#include "velox/functions/sparksql/aggregates/Register.h"
+
+using namespace facebook::velox::exec;
+using namespace facebook::velox::exec::test;
+using namespace facebook::velox::functions::aggregate::test;
+
+namespace facebook::velox::functions::aggregate::sparksql::test {
+
+namespace {
+
+class ArrayAggTest : public AggregationTestBase {
+ protected:
+  static void SetUpTestCase() {
+    functions::aggregate::sparksql::registerAggregateFunctions("");
+  }
+};
+
+TEST_F(ArrayAggTest, sortedGroupBy) {
+  auto data = makeRowVector({
+      makeFlatVector<int16_t>({1, 1, 2, 2, 1, 2, 1}),
+      makeFlatVector<int64_t>({1, 2, 3, 4, 5, 6, 7}),
+      makeFlatVector<int64_t>({10, 20, 30, 40, 50, 60, 70}),
+      makeFlatVector<int32_t>({11, 44, 22, 55, 33, 66, 77}),
+  });
+
+  createDuckDbTable({data});
+
+  // Sorted aggregations over same inputs.
+  auto plan = PlanBuilder()
+                  .values({data})
+                  .singleAggregation(
+                      {"c0"},
+                      {
+                          "array_agg(c1 ORDER BY c2 DESC)",
+                          "array_agg(c1 ORDER BY c3)",
+                      })
+                  .planNode();
+
+  AssertQueryBuilder(plan, duckDbQueryRunner_)
+      .assertResults(
+          "SELECT c0, array_agg(c1 ORDER BY c2 DESC), array_agg(c1 ORDER BY c3) "
+          " FROM tmp GROUP BY 1");
+
+  // Sorted aggregations over different inputs.
+  plan = PlanBuilder()
+             .values({data})
+             .singleAggregation(
+                 {"c0"},
+                 {
+                     "array_agg(c1 ORDER BY c2 DESC)",
+                     "array_agg(c2 ORDER BY c3)",
+                 })
+             .planNode();
+
+  AssertQueryBuilder(plan, duckDbQueryRunner_)
+      .assertResults(
+          "SELECT c0, array_agg(c1 ORDER BY c2 DESC), array_agg(c2 ORDER BY c3) "
+          " FROM tmp GROUP BY 1");
+
+  // Sorted aggregation with multiple sorting keys.
+  plan = PlanBuilder()
+             .values({data})
+             .singleAggregation(
+                 {"c0"},
+                 {
+                     "array_agg(c1 ORDER BY c2 DESC, c3)",
+                 })
+             .planNode();
+
+  AssertQueryBuilder(plan, duckDbQueryRunner_)
+      .assertResults(
+          "SELECT c0, array_agg(c1 ORDER BY c2 DESC, c3) "
+          " FROM tmp GROUP BY 1");
+}
+
+TEST_F(ArrayAggTest, global) {
+  vector_size_t size = 10;
+
+  std::vector<RowVectorPtr> vectors = {makeRowVector({makeFlatVector<int32_t>(
+      size, [](vector_size_t row) { return row * 2; })})};
+
+  createDuckDbTable(vectors);
+  testAggregations(
+      vectors, {}, {"array_agg(c0)"}, "SELECT array_agg(c0) FROM tmp");
+}
+
+TEST_F(ArrayAggTest, globalWithNullData) {
+  auto data = makeRowVector({makeNullableFlatVector<int64_t>(
+      {1, std::nullopt, 3, 4, std::nullopt, 6, 7})});
+  auto expectedResult =
+      makeRowVector({makeArrayVector<int64_t>({{1, 3, 4, 6, 7}})});
+
+  testAggregations({data}, {}, {"array_agg(c0)"}, {expectedResult});
+
+  data = makeRowVector({makeNullableFlatVector<int64_t>(
+      {std::nullopt, 33, std::nullopt, std::nullopt, 66, std::nullopt, 77})});
+  expectedResult = makeRowVector({makeArrayVector<int64_t>({{33, 66, 77}})});
+  testAggregations({data}, {}, {"array_agg(c0)"}, {expectedResult});
+}
+
+TEST_F(ArrayAggTest, globalNoData) {
+  auto data = makeRowVector(ROW({"c0"}, {INTEGER()}), 0);
+  auto expectedResult = makeRowVector({makeArrayVector<int32_t>({{}})});
+  testAggregations({data}, {}, {"array_agg(c0)"}, {expectedResult});
+}
+
+TEST_F(ArrayAggTest, sortedGlobal) {
+  auto data = makeRowVector({
+      makeFlatVector<int64_t>({1, 2, 3, 4, 5, 6, 7}),
+      makeFlatVector<int64_t>({10, 20, 30, 40, 50, 60, 70}),
+      makeFlatVector<int32_t>({11, 33, 22, 44, 66, 55, 77}),
+  });
+
+  createDuckDbTable({data});
+
+  auto plan = PlanBuilder()
+                  .values({data})
+                  .singleAggregation(
+                      {},
+                      {
+                          "array_agg(c0 ORDER BY c1 DESC)",
+                          "array_agg(c0 ORDER BY c2)",
+                      })
+                  .planNode();
+
+  AssertQueryBuilder(plan, duckDbQueryRunner_)
+      .assertResults(
+          "SELECT array_agg(c0 ORDER BY c1 DESC), array_agg(c0 ORDER BY c2) FROM tmp");
+
+  // Sorted aggregations over different inputs.
+  plan = PlanBuilder()
+             .values({data})
+             .singleAggregation(
+                 {},
+                 {
+                     "array_agg(c0 ORDER BY c1 DESC)",
+                     "array_agg(c1 ORDER BY c2)",
+                 })
+             .planNode();
+
+  AssertQueryBuilder(plan, duckDbQueryRunner_)
+      .assertResults(
+          "SELECT array_agg(c0 ORDER BY c1 DESC), array_agg(c1 ORDER BY c2) FROM tmp");
+
+  // Sorted aggregation with multiple sorting keys.
+  plan = PlanBuilder()
+             .values({data})
+             .singleAggregation(
+                 {},
+                 {
+                     "array_agg(c0 ORDER BY c1 DESC, c2)",
+                 })
+             .planNode();
+
+  AssertQueryBuilder(plan, duckDbQueryRunner_)
+      .assertResults("SELECT array_agg(c0 ORDER BY c1 DESC, c2) FROM tmp");
+}
+} // namespace
+} // namespace facebook::velox::functions::aggregate::sparksql::test
diff --git a/velox/functions/sparksql/aggregates/tests/CMakeLists.txt b/velox/functions/sparksql/aggregates/tests/CMakeLists.txt
index 931c56d676577..18b99d790494a 100644
--- a/velox/functions/sparksql/aggregates/tests/CMakeLists.txt
+++ b/velox/functions/sparksql/aggregates/tests/CMakeLists.txt
@@ -14,8 +14,9 @@
 
 add_executable(
   velox_functions_spark_aggregates_test
+  ArrayAggTest.cpp
   BitwiseXorAggregationTest.cpp BloomFilterAggAggregateTest.cpp
-  FirstAggregateTest.cpp LastAggregateTest.cpp Main.cpp)
+  FirstAggregateTest.cpp LastAggregateTest.cpp Main.cpp SetAggTest.cpp)
 
 add_test(velox_functions_spark_aggregates_test
          velox_functions_spark_aggregates_test)
@@ -28,4 +29,6 @@ target_link_libraries(
   velox_functions_spark_aggregates
   gflags::gflags
   gtest
-  gtest_main)
+  gtest_main
+  velox_vector_fuzzer
+  velox_functions_spark)
diff --git a/velox/functions/sparksql/aggregates/tests/SetAggTest.cpp b/velox/functions/sparksql/aggregates/tests/SetAggTest.cpp
new file mode 100644
index 0000000000000..f075c6d233a23
--- /dev/null
+++ b/velox/functions/sparksql/aggregates/tests/SetAggTest.cpp
@@ -0,0 +1,259 @@
+/*
+ * Copyright (c) Facebook, Inc. and its affiliates.
+ *
+ * Licensed 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.
+ */
+#include "velox/common/base/tests/GTestUtils.h"
+#include "velox/exec/tests/utils/AssertQueryBuilder.h"
+#include "velox/functions/lib/aggregates/tests/AggregationTestBase.h"
+#include "velox/functions/sparksql/Register.h"
+#include "velox/functions/sparksql/aggregates/Register.h"
+#include "velox/vector/fuzzer/VectorFuzzer.h"
+
+using namespace facebook::velox::exec;
+using namespace facebook::velox::functions::aggregate::test;
+
+namespace facebook::velox::functions::aggregate::sparksql::test {
+
+namespace {
+
+class SetAggTest : public AggregationTestBase {
+ protected:
+  static void SetUpTestCase() {
+    functions::aggregate::sparksql::registerAggregateFunctions("");
+    facebook::velox::functions::sparksql::registerFunctions("");
+  }
+};
+
+TEST_F(SetAggTest, global) {
+  auto data = makeRowVector({
+      makeFlatVector<int32_t>({1, 2, 3, 4, 5, 3, 4, 2, 6, 7}),
+  });
+
+  auto expected = makeRowVector({
+      makeArrayVector<int32_t>({
+          {1, 2, 3, 4, 5, 6, 7},
+      }),
+  });
+
+  testAggregations({data}, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+
+  // Null inputs.
+  data = makeRowVector({
+      makeNullableFlatVector<int32_t>(
+          {1, 2, std::nullopt, 4, 5, std::nullopt, 4, 2, 6, 7}),
+  });
+
+  expected = makeRowVector({
+      makeArrayVector<int32_t>({
+          {1, 2, 4, 5, 6, 7},
+      }),
+  });
+
+  testAggregations({data}, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+
+  // All inputs are null.
+  data = makeRowVector({
+      makeAllNullFlatVector<int32_t>(1'000),
+  });
+
+  // Empty output.
+  expected = makeRowVector({
+      makeArrayVector<int32_t>({{}}),
+  });
+
+  testAggregations({data}, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+}
+
+TEST_F(SetAggTest, groupBy) {
+  auto data = makeRowVector({
+      makeFlatVector<int16_t>({1, 1, 2, 2, 2, 1, 2, 1, 2, 1}),
+      makeFlatVector<int32_t>({1, 2, 3, 4, 5, 3, 4, 2, 6, 7}),
+  });
+
+  auto expected = makeRowVector({
+      makeFlatVector<int16_t>({1, 2}),
+      makeArrayVector<int32_t>({
+          {1, 2, 3, 7},
+          {3, 4, 5, 6},
+      }),
+  });
+
+  testAggregations(
+      {data}, {"c0"}, {"set_agg(c1)"}, {"c0", "array_sort(a0)"}, {expected});
+
+  // Null inputs.
+  data = makeRowVector({
+      makeFlatVector<int16_t>({1, 1, 2, 2, 2, 1, 2, 1, 2, 1}),
+      makeNullableFlatVector<int32_t>(
+          {1,
+           std::nullopt,
+           3,
+           std::nullopt,
+           5,
+           std::nullopt,
+           3,
+           std::nullopt,
+           6,
+           1}),
+  });
+
+  expected = makeRowVector({
+      makeFlatVector<int16_t>({1, 2}),
+      makeArrayVector<int32_t>({
+          {1},
+          {3, 5, 6},
+      }),
+  });
+
+  testAggregations(
+      {data}, {"c0"}, {"set_agg(c1)"}, {"c0", "array_sort(a0)"}, {expected});
+
+  // All inputs are null for a group.
+  data = makeRowVector({
+      makeFlatVector<int16_t>({1, 1, 2, 2, 2, 1, 2, 1, 2, 1}),
+      makeNullableFlatVector<int32_t>(
+          {1,
+           std::nullopt,
+           std::nullopt,
+           std::nullopt,
+           std::nullopt,
+           std::nullopt,
+           std::nullopt,
+           std::nullopt,
+           std::nullopt,
+           1}),
+  });
+
+  expected = makeRowVector({
+      makeFlatVector<int16_t>({1, 2}),
+      makeArrayVector<int32_t>({
+          {1},
+          {},
+      }),
+  });
+
+  testAggregations(
+      {data}, {"c0"}, {"set_agg(c1)"}, {"c0", "array_sort(a0)"}, {expected});
+}
+
+std::vector<std::optional<std::string>> generateStrings(
+    const std::vector<std::optional<std::string>>& choices,
+    vector_size_t size) {
+  std::vector<int> indices(size);
+  std::iota(indices.begin(), indices.end(), 0); // 0, 1, 2, 3, ...
+
+  std::mt19937 g(std::random_device{}());
+  std::shuffle(indices.begin(), indices.end(), g);
+
+  std::vector<std::optional<std::string>> strings(size);
+  for (auto i = 0; i < size; ++i) {
+    strings[i] = choices[indices[i] % choices.size()];
+  }
+  return strings;
+}
+
+TEST_F(SetAggTest, globalVarchar) {
+  std::vector<std::optional<std::string>> strings = {
+      "grapes",
+      "oranges",
+      "sweet fruits: apple",
+      "sweet fruits: banana",
+      "sweet fruits: papaya",
+  };
+  auto stringVector = makeNullableFlatVector(generateStrings(strings, 25));
+  std::vector<RowVectorPtr> data = {makeRowVector({stringVector})};
+
+  auto expected = makeRowVector({
+      makeNullableArrayVector(
+          std::vector<std::vector<std::optional<std::string>>>{strings}),
+  });
+  testAggregations(data, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+
+  VectorFuzzer::Options options;
+  VectorFuzzer fuzzer(options, pool());
+  data = {
+      makeRowVector({fuzzer.fuzzDictionary(stringVector, 1'000)}),
+      makeRowVector({fuzzer.fuzzDictionary(stringVector, 1'000)}),
+  };
+
+  testAggregations(data, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+
+  // Some nulls.
+  auto stringsAndNull = strings;
+  stringsAndNull.push_back(std::nullopt);
+
+  auto stringVectorWithNulls =
+      makeNullableFlatVector(generateStrings(stringsAndNull, 25));
+
+  data = {makeRowVector({stringVectorWithNulls})};
+
+  expected = makeRowVector({
+      makeNullableArrayVector(
+          std::vector<std::vector<std::optional<std::string>>>{strings}),
+  });
+
+  testAggregations(data, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+
+  data = {
+      makeRowVector({fuzzer.fuzzDictionary(stringVector, 1'000)}),
+      makeRowVector({fuzzer.fuzzDictionary(stringVectorWithNulls, 1'000)}),
+  };
+
+  testAggregations(data, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+
+  // All inputs are null.
+  data = {makeRowVector({makeAllNullFlatVector<StringView>(1'000)})};
+  expected = makeRowVector({
+      makeArrayVector<std::string>({
+          {},
+      }),
+  });
+  testAggregations(data, {}, {"set_agg(c0)"}, {"array_sort(a0)"}, {expected});
+}
+
+TEST_F(SetAggTest, groupByVarchar) {
+  std::vector<std::string> strings = {
+      "grapes",
+      "oranges",
+      "sweet fruits: apple",
+      "sweet fruits: banana",
+      "sweet fruits: papaya",
+  };
+
+  std::vector<RowVectorPtr> data = {makeRowVector({
+      makeFlatVector<int16_t>({1, 1, 2, 2, 2, 1, 2}),
+      makeFlatVector<std::string>({
+          strings[0],
+          strings[1],
+          strings[2],
+          strings[3],
+          strings[4],
+          strings[0],
+          strings[3],
+      }),
+  })};
+
+  auto expected = makeRowVector({
+      makeFlatVector<int16_t>({1, 2}),
+      makeArrayVector<std::string>({
+          {strings[0], strings[1]},
+          {strings[2], strings[3], strings[4]},
+      }),
+  });
+
+  testAggregations(
+      data, {"c0"}, {"set_agg(c1)"}, {"c0", "array_sort(a0)"}, {expected});
+}
+} // namespace
+} // namespace facebook::velox::functions::aggregate::sparksql::test