fix(layout): stop dict writer looping on an oversized value (#10400)

Fixes #10397

A value larger than the dictionary byte budget did not fit an empty
dictionary. The encoder then encoded 0 rows, and
`DictStreamState::encode` retried the same chunk forever.

The bytes and primitive dictionary builders now always admit the first
entry, so each encode step makes progress.

---------

Signed-off-by: Claude <noreply@anthropic.com>
Signed-off-by: Robert Kruszewski <github@robertk.io>
Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com>
Co-authored-by: Robert Kruszewski <robert3005@users.noreply.github.com>
Co-authored-by: Robert Kruszewski <github@robertk.io>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

Author: 4 people

vortex-array/src/builders/dict/bytes.rs

@@ -130,7 +130,10 @@ impl<Code: UnsignedPType> BytesDictBuilder<Code> {
         ) {
             HashTableEntry::Occupied(occupied) => Some(*occupied.get()),
             HashTableEntry::Vacant(vacant) => {
-                if self.views.len() >= self.max_dict_len {
+                // Admit the first entry even if it exceeds the budget, so the caller makes progress.
+                let is_first_entry = self.views.is_empty();
+
+                if !is_first_entry && self.views.len() >= self.max_dict_len {
                     return None;
                 }
 
@@ -146,7 +149,7 @@ impl<Code: UnsignedPType> BytesDictBuilder<Code> {
                     size_of::<BinaryView>() + val.len()
                 };
 
-                if self.dict_bytes() + additional_bytes > self.max_dict_bytes {
+                if !is_first_entry && self.dict_bytes() + additional_bytes > self.max_dict_bytes {
                     return None;
                 }
 
@@ -171,8 +174,9 @@ impl<Code: UnsignedPType> BytesDictBuilder<Code> {
             return Some(*code);
         }
 
-        if self.views.len() >= self.max_dict_len
-            || self.dict_bytes() + size_of::<BinaryView>() > self.max_dict_bytes
+        if !self.views.is_empty()
+            && (self.views.len() >= self.max_dict_len
+                || self.dict_bytes() + size_of::<BinaryView>() > self.max_dict_bytes)
         {
             return None;
         }

vortex-array/src/builders/dict/primitive.rs

@@ -101,7 +101,8 @@ where
         match self.lookup.entry(NativeValue(v)) {
             Entry::Occupied(o) => Some(*o.get()),
             Entry::Vacant(vac) => {
-                if self.values.len() >= self.max_dict_len {
+                // Admit the first entry even if it exceeds the budget, so the caller makes progress.
+                if !self.values.is_empty() && self.values.len() >= self.max_dict_len {
                     return None;
                 }
                 let next_code = Code::from_usize(self.values.len()).unwrap_or_else(|| {
@@ -121,7 +122,7 @@ where
             return Some(*code);
         }
 
-        if self.values.len() >= self.max_dict_len {
+        if !self.values.is_empty() && self.values.len() >= self.max_dict_len {
             return None;
         }
 

vortex-layout/src/layouts/dict/writer.rs

@@ -597,8 +597,12 @@ mod tests {
     use vortex_array::dtype::Nullability::NonNullable;
     use vortex_array::dtype::PType;
     use vortex_array::session::ArraySession;
+    use vortex_error::VortexResult;
     use vortex_session::VortexSession;
 
+    use super::DictChunkLabeler;
+    use super::DictStreamState;
+    use super::DictionaryChunk;
     use super::DictionaryTransformer;
     use super::dict_encode_stream;
     use crate::sequence::SequenceId;
@@ -695,4 +699,55 @@ mod tests {
             "codes stream should use U16 dtype for dictionaries with >255 entries"
         );
     }
+
+    fn encoded_row_count(chunks: &[DictionaryChunk]) -> usize {
+        chunks
+            .iter()
+            .map(|chunk| match chunk {
+                DictionaryChunk::Codes { codes, .. } => codes.len(),
+                DictionaryChunk::Values(_) => 0,
+            })
+            .sum()
+    }
+
+    /// A 49-byte value plus its 16-byte `BinaryView` exceeds the 64-byte budget of an empty
+    /// dictionary. Encoding must still make progress instead of looping forever.
+    #[test]
+    fn test_encode_value_larger_than_dictionary_budget_terminates() -> VortexResult<()> {
+        let mut state = DictStreamState {
+            encoder: None,
+            constraints: DictConstraints {
+                max_bytes: 64,
+                max_len: 100,
+            },
+        };
+
+        let value = "x".repeat(49);
+        let chunk = VarBinArray::from(vec![value.as_str()]).into_array();
+        let mut labeler = DictChunkLabeler::new(SequenceId::root().advance());
+        let encoded = state.encode(&mut labeler, chunk, &mut SESSION.create_execution_ctx())?;
+
+        assert_eq!(encoded_row_count(&encoded), 1);
+        Ok(())
+    }
+
+    #[test]
+    fn test_encode_splits_oversized_value_into_its_own_dictionary() -> VortexResult<()> {
+        let mut state = DictStreamState {
+            encoder: None,
+            constraints: DictConstraints {
+                max_bytes: 64,
+                max_len: 100,
+            },
+        };
+
+        let big = "x".repeat(49);
+        let chunk = VarBinArray::from(vec!["a", big.as_str(), "b"]).into_array();
+        let mut labeler = DictChunkLabeler::new(SequenceId::root().advance());
+        let encoded = state.encode(&mut labeler, chunk, &mut SESSION.create_execution_ctx())?;
+        let drained = state.drain_values(&mut labeler);
+
+        assert_eq!(encoded_row_count(&encoded) + encoded_row_count(&drained), 3);
+        Ok(())
+    }
 }