diff --git a/src/daft-core/src/array/ops/full.rs b/src/daft-core/src/array/ops/full.rs
index f2f86c5298..153a3e0c72 100644
--- a/src/daft-core/src/array/ops/full.rs
+++ b/src/daft-core/src/array/ops/full.rs
@@ -76,7 +76,11 @@ where
     <L::PhysicalType as DaftDataType>::ArrayType: FullNull,
 {
     fn full_null(name: &str, dtype: &DataType, length: usize) -> Self {
-        let physical = <L::PhysicalType as DaftDataType>::ArrayType::full_null(name, dtype, length);
+        let physical = <L::PhysicalType as DaftDataType>::ArrayType::full_null(
+            name,
+            &dtype.to_physical(),
+            length,
+        );
         Self::new(Field::new(name, dtype.clone()), physical)
     }
 
diff --git a/src/daft-core/src/array/ops/image.rs b/src/daft-core/src/array/ops/image.rs
index 5b346b4842..3911165a2c 100644
--- a/src/daft-core/src/array/ops/image.rs
+++ b/src/daft-core/src/array/ops/image.rs
@@ -407,11 +407,7 @@ impl ImageArray {
         sidecar_data: ImageArraySidecarData,
     ) -> DaftResult<Self> {
         if data.is_empty() {
-            // Create an all-null array if the data array is empty.
-            return Ok(ImageArray::new(
-                Field::new(name, data_type.clone()),
-                StructArray::empty(name, &data_type.to_physical()),
-            ));
+            return Ok(ImageArray::full_null(name, &data_type, offsets.len() - 1));
         }
         let offsets = arrow2::offset::OffsetsBuffer::try_from(offsets)?;
         let arrow_dtype: arrow2::datatypes::DataType = T::PRIMITIVE.into();
diff --git a/src/daft-core/src/series/ops/downcast.rs b/src/daft-core/src/series/ops/downcast.rs
index cceb874787..f70efc697f 100644
--- a/src/daft-core/src/series/ops/downcast.rs
+++ b/src/daft-core/src/series/ops/downcast.rs
@@ -7,6 +7,8 @@ use crate::series::array_impl::ArrayWrapper;
 use crate::series::Series;
 use common_error::DaftResult;
 
+use self::logical::ImageArray;
+
 impl Series {
     pub fn downcast<Arr: DaftArrayType>(&self) -> DaftResult<&Arr> {
         match self.inner.as_any().downcast_ref() {
@@ -97,6 +99,10 @@ impl Series {
         self.downcast()
     }
 
+    pub fn image(&self) -> DaftResult<&ImageArray> {
+        self.downcast()
+    }
+
     pub fn fixed_size_image(&self) -> DaftResult<&FixedShapeImageArray> {
         self.downcast()
     }
diff --git a/tests/table/image/test_decode.py b/tests/table/image/test_decode.py
new file mode 100644
index 0000000000..9cd7bd3a3e
--- /dev/null
+++ b/tests/table/image/test_decode.py
@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+import daft
+
+
+def test_decode_all_empty():
+    df = daft.from_pydict({"foo": [b"not an image", None]})
+    df = df.with_column("image", df["foo"].image.decode(on_error="null"))
+    df.collect()
+
+    assert df.to_pydict() == {
+        "foo": [b"not an image", None],
+        "image": [None, None],
+    }