1use arrow_array::ArrayRef as ArrowArrayRef;
7use arrow_schema::Field;
8use arrow_schema::extension::ExtensionType;
9use arrow_schema::extension::Json as ArrowJson;
10use vortex_array::ArrayRef;
11use vortex_array::EmptyMetadata;
12use vortex_array::ExecutionCtx;
13use vortex_array::IntoArray;
14use vortex_array::arrays::ExtensionArray;
15use vortex_array::arrays::extension::ExtensionArrayExt;
16use vortex_array::arrow::ArrowExport;
17use vortex_array::arrow::ArrowExportVTable;
18use vortex_array::arrow::ArrowImport;
19use vortex_array::arrow::ArrowImportVTable;
20use vortex_array::arrow::ArrowSession;
21use vortex_array::arrow::ArrowSessionExt;
22use vortex_array::arrow::FromArrowArray;
23use vortex_array::dtype::DType;
24use vortex_array::dtype::extension::ExtDType;
25use vortex_array::dtype::extension::ExtVTable;
26use vortex_error::VortexExpect;
27use vortex_error::VortexResult;
28use vortex_session::registry::CachedId;
29use vortex_session::registry::Id;
30
31use crate::Json;
32
33static ARROW_JSON: CachedId = CachedId::new(ArrowJson::NAME);
35
36fn has_valid_json_extension(field: &Field) -> bool {
38 field.extension_type_name() == Some(ArrowJson::NAME)
39 && ArrowJson::try_new_from_field_metadata(field.data_type(), field.metadata()).is_ok()
40}
41
42impl ArrowExportVTable for Json {
43 fn arrow_ext_id(&self) -> Id {
44 *ARROW_JSON
45 }
46
47 fn vortex_id(&self) -> Id {
48 Json.id()
49 }
50
51 fn to_arrow_field(
52 &self,
53 name: &str,
54 dtype: &DType,
55 session: &ArrowSession,
56 ) -> VortexResult<Option<Field>> {
57 let DType::Extension(ext_dtype) = dtype else {
58 return Ok(None);
59 };
60 if !ext_dtype.is::<Json>() {
61 return Ok(None);
62 }
63
64 let mut field = session.to_arrow_field(name, ext_dtype.storage_dtype())?;
65 field
66 .try_with_extension_type(ArrowJson::default())
67 .vortex_expect("Utf8 is a valid storage type for Arrow JSON");
68 Ok(Some(field))
69 }
70
71 fn execute_arrow(
72 &self,
73 array: ArrayRef,
74 target: &Field,
75 ctx: &mut ExecutionCtx,
76 ) -> VortexResult<ArrowExport> {
77 let is_json = array
78 .dtype()
79 .as_extension_opt()
80 .map(|ext_dtype| ext_dtype.is::<Json>())
81 .unwrap_or(false);
82 if !is_json {
83 return Ok(ArrowExport::Unsupported(array));
84 }
85
86 ArrowJson::try_new_from_field_metadata(target.data_type(), target.metadata())?;
87
88 let executed = array.execute::<ExtensionArray>(ctx)?;
89 let storage = executed.storage_array().clone();
90 let storage_field = Field::new(
91 String::new(),
92 target.data_type().clone(),
93 target.is_nullable(),
94 );
95 let session = ctx.session().clone();
96
97 let storage = session
98 .arrow()
99 .execute_arrow(storage, Some(&storage_field), ctx)?;
100
101 Ok(ArrowExport::Exported(storage))
102 }
103}
104
105impl ArrowImportVTable for Json {
106 fn arrow_ext_id(&self) -> Id {
107 *ARROW_JSON
108 }
109
110 fn from_arrow_field(&self, field: &Field) -> VortexResult<Option<DType>> {
111 if !has_valid_json_extension(field) {
112 return Ok(None);
113 }
114
115 Ok(Some(DType::Extension(
116 ExtDType::<Json>::try_new(EmptyMetadata, DType::Utf8(field.is_nullable().into()))?
117 .erased(),
118 )))
119 }
120
121 fn from_arrow_array(
122 &self,
123 array: ArrowArrayRef,
124 field: &Field,
125 dtype: &DType,
126 ) -> VortexResult<ArrowImport> {
127 let DType::Extension(ext_dtype) = dtype else {
128 return Ok(ArrowImport::Unsupported(array));
129 };
130 if !ext_dtype.is::<Json>() || !has_valid_json_extension(field) {
131 return Ok(ArrowImport::Unsupported(array));
132 }
133
134 let storage = ArrayRef::from_arrow(array.as_ref(), field.is_nullable())?;
135 Ok(ArrowImport::Imported(
136 ExtensionArray::new(ext_dtype.clone(), storage).into_array(),
137 ))
138 }
139}
140
141#[cfg(test)]
142mod tests {
143
144 use std::sync::Arc;
145
146 use arrow_array::Array;
147 use arrow_array::ArrayRef as ArrowArrayRef;
148 use arrow_array::StringArray;
149 use arrow_array::cast::AsArray;
150 use arrow_schema::DataType;
151 use arrow_schema::Field;
152 use arrow_schema::extension::ExtensionType;
153 use arrow_schema::extension::Json as ArrowJson;
154 use vortex_array::EmptyMetadata;
155 use vortex_array::IntoArray;
156 use vortex_array::VortexSessionExecute;
157 use vortex_array::arrays::ExtensionArray;
158 use vortex_array::arrays::VarBinArray;
159 use vortex_array::arrow::ArrowSessionExt;
160 use vortex_array::dtype::Nullability;
161 use vortex_array::dtype::extension::ExtDType;
162 use vortex_error::VortexExpect;
163 use vortex_error::VortexResult;
164
165 use crate::Json;
166 use crate::initialize;
167
168 #[test]
170 fn exports_json_extension_array_as_arrow_json() -> VortexResult<()> {
171 let session = vortex_array::array_session();
172 initialize(&session);
173
174 let storage = VarBinArray::from_iter(
175 [Some("{\"id\":1}"), Some("{\"id\":2}")],
176 vortex_array::dtype::DType::Utf8(Nullability::NonNullable),
177 )
178 .into_array();
179 let ext_dtype = ExtDType::<Json>::try_new(EmptyMetadata, storage.dtype().clone())?.erased();
180
181 dbg!(&ext_dtype);
182 let array = ExtensionArray::new(ext_dtype, storage).into_array();
183
184 let field = session.arrow().to_arrow_field("data", array.dtype())?;
185 assert_eq!(field.extension_type_name(), Some(ArrowJson::NAME));
186 ArrowJson::try_new_from_field_metadata(field.data_type(), field.metadata())?;
187
188 dbg!(&field);
189
190 let exported = session.arrow().execute_arrow(
191 array,
192 Some(&field),
193 &mut session.create_execution_ctx(),
194 )?;
195
196 assert!(exported.data_type().is_string());
197
198 dbg!(exported.data_type());
199
200 let strings = exported.as_string_view();
201 assert_eq!(strings.value(0), "{\"id\":1}");
202 assert_eq!(strings.value(1), "{\"id\":2}");
203 Ok(())
204 }
205
206 #[test]
208 fn imports_arrow_json_extension_array_as_vortex_json() -> VortexResult<()> {
209 let session = vortex_array::array_session();
210 initialize(&session);
211
212 let mut field = Field::new("data", DataType::Utf8, false);
213 field.try_with_extension_type(ArrowJson::default())?;
214 let array = Arc::new(StringArray::from(vec!["{\"id\":1}", "{\"id\":2}"])) as ArrowArrayRef;
215
216 let imported = session.arrow().from_arrow_array(array, &field)?;
217 let ext_dtype = imported
218 .dtype()
219 .as_extension_opt()
220 .vortex_expect("expected JSON extension dtype");
221 assert!(ext_dtype.is::<Json>());
222
223 let exported = session.arrow().execute_arrow(
224 imported,
225 Some(&field),
226 &mut session.create_execution_ctx(),
227 )?;
228 let strings = exported.as_string::<i32>();
229 assert_eq!(strings.value(0), "{\"id\":1}");
230 assert_eq!(strings.value(1), "{\"id\":2}");
231 Ok(())
232 }
233}