added two stage lookup

Author: savs02

arrow-pg/src/encoder.rs

@@ -522,13 +522,37 @@ pub fn encode_value<T: Encoder>(
             if arr.is_null(idx) {
                 return encoder.encode_field_with_type_and_format(&None::<i8>, type_, format);
             }
+
+            macro_rules! get_value_index {
+                ($key_ty:ty) => {
+                    arr.as_any()
+                        .downcast_ref::<DictionaryArray<$key_ty>>()
+                        .map(|dict| dict.keys().value(idx) as usize)
+                };
+            }
+
+            let value_idx = get_value_index!(Int8Type)
+                .or_else(|| get_value_index!(Int16Type))
+                .or_else(|| get_value_index!(Int32Type))
+                .or_else(|| get_value_index!(Int64Type))
+                .or_else(|| get_value_index!(UInt8Type))
+                .or_else(|| get_value_index!(UInt16Type))
+                .or_else(|| get_value_index!(UInt32Type))
+                .or_else(|| get_value_index!(UInt64Type))
+                .ok_or_else(|| {
+                    ToSqlError::from(format!(
+                        "Unsupported dictionary key type"
+                    ))
+                })?;
+
+
             // Get the dictionary values, ignoring keys
             // We'll use Int32Type as a common key type, but we're only interested in values
             macro_rules! get_dict_values {
                 ($key_type:ty) => {
                     arr.as_any()
                         .downcast_ref::<DictionaryArray<$key_type>>()
-                        .map(|dict| dict.values())
+                        .map(|dict| dict.values().clone())
                 };
             }
 
@@ -543,16 +567,16 @@ pub fn encode_value<T: Encoder>(
                 .or_else(|| get_dict_values!(UInt64Type))
                 .ok_or_else(|| {
                     ToSqlError::from(format!(
-                        "Unsupported dictionary key type for value type {value_type}"
+                        "Unsupported dictionary key type"
                     ))
                 })?;
 
             // If the dictionary has only one value, treat it as a primitive
             if values.len() == 1 {
-                encode_value(encoder, values, 0, type_, format)?
+                encode_value(encoder, &values, 0, type_, format)?
             } else {
                 // Otherwise, use value directly indexed by values array
-                encode_value(encoder, values, idx, type_, format)?
+                encode_value(encoder, &values, idx, type_, format)?
             }
         }
         _ => {