Fix UBSan misaligned AggregateState access in hash aggregate scan

src/common/in_mem_overflow_buffer.cpp

@@ -23,6 +23,14 @@ uint8_t* BufferBlock::data() const {
 }
 
 uint8_t* InMemOverflowBuffer::allocateSpace(uint64_t size) {
+    if (!blocks.empty()) {
+        // Keep every allocation 8-byte aligned: consumers store pointers and uint64s in
+        // overflow memory (e.g. COLLECT/HISTOGRAM linked-list elements), and an unaligned
+        // bump offset would make those accesses UB (UBSan: "store to misaligned address").
+        static constexpr uint64_t OVERFLOW_ALIGNMENT = 8;
+        currentBlock()->currentOffset =
+            (currentBlock()->currentOffset + OVERFLOW_ALIGNMENT - 1) & ~(OVERFLOW_ALIGNMENT - 1);
+    }
     if (requireNewBlock(size)) {
         if (!blocks.empty() && currentBlock()->currentOffset == 0) {
             blocks.pop_back();

src/function/find_function.cpp

@@ -47,9 +47,14 @@ int64_t Find::alignedNeedleSizeFind(const uint8_t* haystack, uint32_t haystackLe
     if (sizeof(UNSIGNED) > haystackLen) {
         return -1;
     }
-    auto needleVal = *((UNSIGNED*)needle);
+    // haystack + offset is inherently misaligned for most offsets (and needle itself comes
+    // from an unaligned string buffer), so compare through memcpy'd locals instead of
+    // dereferencing possibly-misaligned pointers (UBSan: "load of misaligned address").
+    UNSIGNED needleVal;
+    memcpy(&needleVal, needle, sizeof(UNSIGNED));
     for (auto offset = 0u; offset <= haystackLen - sizeof(UNSIGNED); offset++) {
-        auto haystackVal = *((UNSIGNED*)(haystack + offset));
+        UNSIGNED haystackVal;
+        memcpy(&haystackVal, haystack + offset, sizeof(UNSIGNED));
         if (needleVal == haystackVal) {
             return firstMatchCharOffset + offset;
         }

src/include/function/aggregate_function.h

@@ -1,7 +1,11 @@
 #pragma once
 
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
 #include <functional>
 #include <utility>
+#include <vector>
 
 #include "common/in_mem_overflow_buffer.h"
 #include "common/vector/value_vector.h"
@@ -34,6 +38,37 @@ using aggr_combine_function_t = std::function<void(uint8_t* state, uint8_t* othe
     common::InMemOverflowBuffer* overflowBuffer)>;
 using aggr_finalize_function_t = std::function<void(uint8_t* state)>;
 
+namespace detail {
+// Aggregate states stored in packed factorized-table tuples may be misaligned for their
+// concrete state type (tuples have no alignment padding). The per-function update/combine/
+// finalize implementations reinterpret_cast the raw bytes and touch members, which is UB on
+// a misaligned pointer (UBSan: "member access within misaligned address"). Route every call
+// through an aligned staging buffer: states that already satisfy the strictest fundamental
+// alignment go straight through (the common case, e.g. heap-allocated simple-aggregate
+// states); the rest are copied in with memcpy, executed on the aligned copy, and copied
+// back. Two slots are provided so combine() can stage both inputs. See also
+// copyStateToAlignedBuffer in processor/map/map_aggregate.cpp for the scan path.
+inline constexpr size_t AGG_STATE_ALIGNMENT = alignof(std::max_align_t);
+
+inline bool isAggregateStateAligned(const uint8_t* state) {
+    return reinterpret_cast<uintptr_t>(state) % AGG_STATE_ALIGNMENT == 0;
+}
+
+inline std::vector<uint8_t>& stagedAggregateStateBuffer(int slot) {
+    thread_local std::vector<uint8_t> buffers[2];
+    return buffers[slot];
+}
+
+inline uint8_t* stageAggregateState(const uint8_t* state, uint32_t size, int slot) {
+    auto& buffer = stagedAggregateStateBuffer(slot);
+    if (buffer.size() < size) {
+        buffer.resize(size);
+    }
+    memcpy(buffer.data(), state, size);
+    return buffer.data();
+}
+} // namespace detail
+
 struct AggregateFunction final : public ScalarOrAggregateFunction {
     bool isDistinct;
     bool needToHandleNulls = false;
@@ -75,20 +110,47 @@ struct AggregateFunction final : public ScalarOrAggregateFunction {
 
     void updateAllState(uint8_t* state, common::ValueVector* input, uint64_t multiplicity,
         common::InMemOverflowBuffer* overflowBuffer) const {
-        return updateAllFunc(state, input, multiplicity, overflowBuffer);
+        if (detail::isAggregateStateAligned(state)) {
+            return updateAllFunc(state, input, multiplicity, overflowBuffer);
+        }
+        auto size = static_cast<uint32_t>(getAggregateStateSize());
+        auto* staged = detail::stageAggregateState(state, size, 0);
+        updateAllFunc(staged, input, multiplicity, overflowBuffer);
+        memcpy(state, staged, size);
     }
 
     void updatePosState(uint8_t* state, common::ValueVector* input, uint64_t multiplicity,
         uint32_t pos, common::InMemOverflowBuffer* overflowBuffer) const {
-        return updatePosFunc(state, input, multiplicity, pos, overflowBuffer);
+        if (detail::isAggregateStateAligned(state)) {
+            return updatePosFunc(state, input, multiplicity, pos, overflowBuffer);
+        }
+        auto size = static_cast<uint32_t>(getAggregateStateSize());
+        auto* staged = detail::stageAggregateState(state, size, 0);
+        updatePosFunc(staged, input, multiplicity, pos, overflowBuffer);
+        memcpy(state, staged, size);
     }
 
     void combineState(uint8_t* state, uint8_t* otherState,
         common::InMemOverflowBuffer* overflowBuffer) const {
-        return combineFunc(state, otherState, overflowBuffer);
+        if (detail::isAggregateStateAligned(state) && detail::isAggregateStateAligned(otherState)) {
+            return combineFunc(state, otherState, overflowBuffer);
+        }
+        auto size = static_cast<uint32_t>(getAggregateStateSize());
+        auto* staged = detail::stageAggregateState(state, size, 0);
+        auto* stagedOther = detail::stageAggregateState(otherState, size, 1);
+        combineFunc(staged, stagedOther, overflowBuffer);
+        memcpy(state, staged, size);
     }
 
-    void finalizeState(uint8_t* state) const { return finalizeFunc(state); }
+    void finalizeState(uint8_t* state) const {
+        if (detail::isAggregateStateAligned(state)) {
+            return finalizeFunc(state);
+        }
+        auto size = static_cast<uint32_t>(getAggregateStateSize());
+        auto* staged = detail::stageAggregateState(state, size, 0);
+        finalizeFunc(staged);
+        memcpy(state, staged, size);
+    }
 
     bool isFunctionDistinct() const { return isDistinct; }
 

src/processor/map/map_aggregate.cpp

@@ -1,4 +1,6 @@
 #include <algorithm>
+#include <cstring>
+#include <vector>
 
 #include "binder/expression/aggregate_function_expression.h"
 #include "binder/expression/literal_expression.h"
@@ -91,29 +93,55 @@ static std::vector<AggregateFunction> getAggFunctions(const expression_vector& a
     return aggregateFunctions;
 }
 
+// Aggregate states are stored in packed factorized-table tuples without alignment padding,
+// so a state pointer derived from a tuple may be misaligned for AggregateState (which
+// requires 8-byte alignment because of its vptr). Virtual dispatch on such a pointer -- and
+// member access inside the callee -- is UB, reported by UBSan as "member access within
+// misaligned address". Copy the state bytes into an aligned thread-local buffer first
+// (memcpy has no alignment requirement) and dispatch on the aligned copy. The copy carries
+// the vptr bit pattern, so the dynamic type is preserved. Thread-local storage keeps the
+// buffer reusable across calls without data races between scan threads.
+static AggregateState* copyStateToAlignedBuffer(AggregateState* aggregateState,
+    uint32_t stateSize) {
+    thread_local std::vector<uint8_t> alignedBuffer;
+    if (alignedBuffer.size() < stateSize) {
+        alignedBuffer.resize(stateSize);
+    }
+    memcpy(alignedBuffer.data(), aggregateState, stateSize);
+    return reinterpret_cast<AggregateState*>(alignedBuffer.data());
+}
+
 static void writeAggResultWithNullToVector(ValueVector& vector, uint64_t pos,
-    AggregateState* aggregateState) {
-    auto isNull = aggregateState->constCast<AggregateStateWithNull>().isNull;
+    AggregateState* aggregateState, uint32_t stateSize) {
+    auto* alignedState = copyStateToAlignedBuffer(aggregateState, stateSize);
+    auto isNull = alignedState->constCast<AggregateStateWithNull>().isNull;
     vector.setNull(pos, isNull);
     if (!isNull) {
-        aggregateState->writeToVector(&vector, pos);
+        alignedState->writeToVector(&vector, pos);
     }
 }
 
 static void writeAggResultWithoutNullToVector(ValueVector& vector, uint64_t pos,
-    AggregateState* aggregateState) {
+    AggregateState* aggregateState, uint32_t stateSize) {
     vector.setNull(pos, false);
-    aggregateState->writeToVector(&vector, pos);
+    copyStateToAlignedBuffer(aggregateState, stateSize)->writeToVector(&vector, pos);
 }
 
 static std::vector<move_agg_result_to_vector_func> getMoveAggResultToVectorFuncs(
     std::vector<AggregateFunction>& aggregateFunctions) {
     std::vector<move_agg_result_to_vector_func> moveAggResultToVectorFuncs;
     for (auto& aggregateFunction : aggregateFunctions) {
+        auto stateSize = static_cast<uint32_t>(aggregateFunction.getAggregateStateSize());
         if (aggregateFunction.needToHandleNulls) {
-            moveAggResultToVectorFuncs.push_back(writeAggResultWithoutNullToVector);
+            moveAggResultToVectorFuncs.push_back(
+                [stateSize](ValueVector& vector, uint64_t pos, AggregateState* aggregateState) {
+                    writeAggResultWithoutNullToVector(vector, pos, aggregateState, stateSize);
+                });
         } else {
-            moveAggResultToVectorFuncs.push_back(writeAggResultWithNullToVector);
+            moveAggResultToVectorFuncs.push_back(
+                [stateSize](ValueVector& vector, uint64_t pos, AggregateState* aggregateState) {
+                    writeAggResultWithNullToVector(vector, pos, aggregateState, stateSize);
+                });
         }
     }
     return moveAggResultToVectorFuncs;

src/processor/operator/aggregate/hash_aggregate_scan.cpp

@@ -24,14 +24,20 @@ bool HashAggregateScan::getNextTuplesInternal(ExecutionContext* /*context*/) {
     entries.resize(numRowsToScan);
     sharedState->scan(entries, groupByKeyVectors, startOffset, numRowsToScan,
         groupByKeyVectorsColIdxes);
+    // Aggregate states are packed without alignment padding, so `entry + offset` may be
+    // misaligned for AggregateState: never call virtuals (e.g. getStateSize()) on it here.
+    // State byte sizes come from the table schema instead (each state column was sized with
+    // AggregateFunction::getAggregateStateSize() at plan time); the move funcs copy each
+    // state to an aligned buffer before dispatching.
+    auto tableSchema = sharedState->getTableSchema();
     for (auto pos = 0u; pos < numRowsToScan; ++pos) {
         auto entry = entries[pos];
-        auto offset = sharedState->getTableSchema()->getColOffset(groupByKeyVectors.size());
+        auto offset = tableSchema->getColOffset(groupByKeyVectors.size());
         for (auto i = 0u; i < aggregateVectors.size(); i++) {
             auto vector = aggregateVectors[i];
             auto aggState = reinterpret_cast<AggregateState*>(entry + offset);
             scanInfo.moveAggResultToVectorFuncs[i](*vector, pos, aggState);
-            offset += aggState->getStateSize();
+            offset += tableSchema->getColumn(groupByKeyVectors.size() + i)->getNumBytes();
         }
     }
     metrics->numOutputTuple.increase(numRowsToScan);

src/processor/operator/persistent/reader/parquet/interval_column_reader.cpp

@@ -1,14 +1,22 @@
 #include "processor/operator/persistent/reader/parquet/interval_column_reader.h"
 
+#include <cstring>
+
 namespace lbug {
 namespace processor {
 
 common::interval_t IntervalValueConversion::readParquetInterval(const char* input) {
+    // The parquet plain-data buffer has no alignment guarantee, so copy each field out with
+    // memcpy instead of dereferencing a possibly-misaligned uint32_t*
+    // (UBSan: "load of misaligned address").
     common::interval_t result;
-    auto inputData = reinterpret_cast<const uint32_t*>(input);
-    result.months = inputData[0];
-    result.days = inputData[1];
-    result.micros = int64_t(inputData[2]) * 1000;
+    uint32_t months, days, microsLow;
+    memcpy(&months, input, sizeof(uint32_t));
+    memcpy(&days, input + sizeof(uint32_t), sizeof(uint32_t));
+    memcpy(&microsLow, input + 2 * sizeof(uint32_t), sizeof(uint32_t));
+    result.months = months;
+    result.days = days;
+    result.micros = int64_t(microsLow) * 1000;
     return result;
 }
 

src/processor/result/base_hash_table.cpp

@@ -1,6 +1,7 @@
 #include "processor/result/base_hash_table.h"
 
 #include <cmath>
+#include <cstring>
 
 #include "common/constants.h"
 #include "common/null_buffer.h"
@@ -79,8 +80,13 @@ static ft_compare_function_t getFactorizedTableCompareEntryFunc(const LogicalTyp
 template<>
 bool factorizedTableCompareEntry<list_entry_t>(const uint8_t* entry1, const uint8_t* entry2,
     const LogicalType& type) {
-    const auto* list1 = reinterpret_cast<const list_t*>(entry1);
-    const auto* list2 = reinterpret_cast<const list_t*>(entry2);
+    // entries point into packed factorized-table tuples without alignment padding, so they
+    // may be misaligned for list_t (8-byte alignment). Copy through aligned temporaries.
+    list_t list1Copy, list2Copy;
+    memcpy(&list1Copy, entry1, sizeof(list_t));
+    memcpy(&list2Copy, entry2, sizeof(list_t));
+    const auto* list1 = &list1Copy;
+    const auto* list2 = &list2Copy;
     if (list1->size != list2->size) {
         return false;
     }
@@ -164,7 +170,12 @@ template<>
     uint32_t vectorPos, const uint8_t* entry) {
     auto dataVector = ListVector::getDataVector(vector);
     auto listToCompare = vector->getValue<list_entry_t>(vectorPos);
-    auto listEntry = reinterpret_cast<const list_t*>(entry);
+    // `entry` points into a packed factorized-table tuple without alignment padding, so it
+    // may be misaligned for list_t (8-byte alignment). Copy through an aligned temporary
+    // (memcpy has no alignment requirement) instead of touching members in place.
+    list_t listEntryCopy;
+    memcpy(&listEntryCopy, entry, sizeof(list_t));
+    auto listEntry = &listEntryCopy;
     auto entryNullBytes = reinterpret_cast<uint8_t*>(listEntry->overflowPtr);
     auto entryValues = entryNullBytes + NullBuffer::getNumBytesForNullValues(listEntry->size);
     auto rowLayoutSize = LogicalTypeUtils::getRowLayoutSize(dataVector->dataType);