fix(hashmap): accept broadcast const iterator vectors

pkg/common/hashmap/iterator.go

@@ -32,7 +32,19 @@ func validateIteratorVectors(
 		return mpool.ErrAllocationAccountInvalid
 	}
 	for _, vec := range vecs {
-		if vec == nil || start > vec.Length() || count > vec.Length()-start {
+		if vec == nil {
+			return mpool.ErrAllocationAccountInvalid
+		}
+		// Const vectors physically store one value and broadcast it across the
+		// caller's logical row range. Hash encoders already handle that contract
+		// by reading row zero, so only require physical storage when rows are read.
+		if vec.IsConst() {
+			if count > 0 && vec.Length() == 0 {
+				return mpool.ErrAllocationAccountInvalid
+			}
+			continue
+		}
+		if start > vec.Length() || count > vec.Length()-start {
 			return mpool.ErrAllocationAccountInvalid
 		}
 	}

pkg/common/hashmap/strhashmap_test.go

@@ -854,14 +854,14 @@ func TestHashMapIteratorsRejectMalformedRowShapes(t *testing.T) {
 		},
 	} {
 		m, iterator := makeIterator()
-		short, err := vector.NewConstFixed(types.T_int32.ToType(), int32(1), 1, mp)
-		require.NoError(t, err)
+		short := vector.NewVec(types.T_int32.ToType())
+		require.NoError(t, vector.AppendFixed(short, int32(1), false, mp))
 		for _, vecs := range [][]*vector.Vector{
 			nil,
 			{nil},
 			{short},
 		} {
-			_, _, err = iterator.Insert(0, 2, vecs)
+			_, _, err := iterator.Insert(0, 2, vecs)
 			require.ErrorIs(t, err, mpool.ErrAllocationAccountInvalid)
 			_, _, err = iterator.Find(1, 1, vecs)
 			require.ErrorIs(t, err, mpool.ErrAllocationAccountInvalid)
@@ -873,6 +873,54 @@ func TestHashMapIteratorsRejectMalformedRowShapes(t *testing.T) {
 	require.Zero(t, mp.CurrNB())
 }
 
+func TestHashMapIteratorsBroadcastConstVectors(t *testing.T) {
+	mp := mpool.MustNewZero()
+	for _, makeIterator := range []func() (HashMap, Iterator){
+		func() (HashMap, Iterator) {
+			m, err := NewIntHashMap(true, mp)
+			require.NoError(t, err)
+			return m, m.NewIterator()
+		},
+		func() (HashMap, Iterator) {
+			m, err := NewStrHashMap(true, mp)
+			require.NoError(t, err)
+			return m, m.NewIterator()
+		},
+	} {
+		m, iterator := makeIterator()
+		constant, err := vector.NewConstFixed(types.T_int32.ToType(), int32(7), 1, mp)
+		require.NoError(t, err)
+
+		values, zValues, err := iterator.Insert(UnitLimit, 2, []*vector.Vector{constant})
+		require.NoError(t, err)
+		require.Equal(t, []uint64{1, 1}, values)
+		require.Equal(t, []int64{1, 1}, zValues)
+		require.Equal(t, uint64(1), m.GroupCount())
+
+		values, zValues, err = iterator.Find(UnitLimit*2, 2, []*vector.Vector{constant})
+		require.NoError(t, err)
+		require.Equal(t, []uint64{1, 1}, values)
+		require.Equal(t, []int64{1, 1}, zValues)
+
+		constantNull := vector.NewConstNull(types.T_int32.ToType(), 1, mp)
+		values, zValues, err = iterator.Insert(UnitLimit*3, 2, []*vector.Vector{constantNull})
+		require.NoError(t, err)
+		require.Equal(t, []uint64{2, 2}, values)
+		require.Equal(t, []int64{1, 1}, zValues)
+		require.Equal(t, uint64(2), m.GroupCount())
+
+		emptyConstant := vector.NewConstNull(types.T_int32.ToType(), 0, mp)
+		_, _, err = iterator.Insert(UnitLimit*4, 1, []*vector.Vector{emptyConstant})
+		require.ErrorIs(t, err, mpool.ErrAllocationAccountInvalid)
+
+		emptyConstant.Free(mp)
+		constantNull.Free(mp)
+		constant.Free(mp)
+		m.Free()
+	}
+	require.Zero(t, mp.CurrNB())
+}
+
 func runFloatHashMapShape(
 	t *testing.T,
 	composite bool,