mito2/compaction/
json2.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::collections::{HashMap, HashSet};
16use std::sync::Arc;
17
18use arrow_schema::extension::ExtensionType;
19use datatypes::arrow::datatypes::{DataType as ArrowDataType, Field, Schema, SchemaRef};
20use datatypes::arrow::record_batch::RecordBatch;
21use datatypes::extension::json::{JSON2_REMAINDER_FIELD_NAME, Json2ExtensionType, JsonMetadata};
22use datatypes::json::{
23    JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS, JsonSettings, JsonTypeHint, TypeHintMismatchPolicy,
24};
25use datatypes::prelude::ConcreteDataType;
26use datatypes::types::json_type::JsonNativeType;
27use datatypes::vectors::json::array::JsonArray;
28use datatypes::vectors::json::json2_physical_data_type;
29use parquet::arrow::parquet_to_arrow_schema;
30use parquet::file::metadata::ParquetMetaData;
31use snafu::{OptionExt, ResultExt, ensure};
32use store_api::metadata::RegionMetadataRef;
33
34use crate::error::{
35    ConvertValueSnafu, DataTypeMismatchSnafu, InvalidRecordBatchSnafu, NewRecordBatchSnafu, Result,
36};
37
38/// Plan for rewriting one JSON2 column to a fixed compaction layout.
39///
40/// A compaction input may contain v1 and v2 SSTs with different physical schemas. This plan is
41/// derived only from the current region metadata and remains fixed while all input batches are
42/// decoded and rewritten. It therefore prevents source-only paths from expanding the output
43/// schema without a bound.
44pub(crate) struct Json2RewritePlan {
45    /// User-defined settings used to encode logical values.
46    logical_settings: JsonSettings,
47    /// Fixed settings used to build the target physical layout.
48    pub(super) target_layout: JsonSettings,
49}
50
51/// JSON2 rewrite plans keyed by logical column name.
52pub(crate) type Json2RewritePlans = HashMap<String, Json2RewritePlan>;
53
54#[derive(Clone)]
55struct Json2LeafPathStats {
56    rows: u64,
57    /// Number of input schemas that explicitly represent this leaf.
58    sources: usize,
59    data_type: JsonNativeType,
60    is_type_conflicted: bool,
61}
62
63/// Builds the JSON2 rewrite plans for a compaction.
64///
65/// Type hints from current region metadata are always retained. Existing explicit dynamic paths
66/// shared by every input SST schema are ranked once to produce a fixed layout; paths that may
67/// reside in a v2 remainder are deliberately not promoted. [`rewrite_json2_batch`] decodes inputs and rewrites
68/// them according to these plans. Non-JSON2 columns are omitted from the returned map.
69///
70/// Returns an error when a JSON2 column has invalid or missing extension metadata, or when its
71/// output layout is not v2. Legacy region metadata is upgraded in memory by the region opener
72/// before compaction reaches this function.
73pub(super) fn collect_json2_rewrite_plans_from_parquet(
74    metadata: &RegionMetadataRef,
75    parquet_metadata: &[Arc<ParquetMetaData>],
76) -> Result<Json2RewritePlans> {
77    let schemas = parquet_metadata
78        .iter()
79        .map(|metadata| {
80            let file = metadata.file_metadata();
81            let schema = parquet_to_arrow_schema(file.schema_descr(), file.key_value_metadata())
82                .map_err(|error| {
83                    InvalidRecordBatchSnafu {
84                        reason: format!("Failed to read compaction input Arrow schema: {error}"),
85                    }
86                    .build()
87                })?;
88            let rows = metadata
89                .row_groups()
90                .iter()
91                .map(|x| x.num_rows())
92                .sum::<i64>() as u64;
93            Ok((Arc::new(schema), rows))
94        })
95        .collect::<Result<Vec<_>>>()?;
96
97    collect_json2_rewrite_plans(metadata, &schemas)
98}
99
100/// Builds JSON2 rewrite plans from existing physical schemas.
101///
102/// Each source row count weights all of its existing explicit leaves. Ordinary paths are never
103/// discovered from the v2 remainder, so this operation cannot unexpectedly promote opaque data.
104pub(crate) fn collect_json2_rewrite_plans(
105    metadata: &RegionMetadataRef,
106    schemas: &[(SchemaRef, u64)],
107) -> Result<Json2RewritePlans> {
108    let json2_columns = metadata
109        .column_metadatas
110        .iter()
111        .filter_map(|x| {
112            x.column_schema
113                .data_type
114                .is_json2()
115                .then_some(&x.column_schema)
116        })
117        .collect::<Vec<_>>();
118
119    let mut plans = HashMap::with_capacity(json2_columns.len());
120    for column in json2_columns {
121        let extension = column
122            .extension_type::<Json2ExtensionType>()
123            .context(DataTypeMismatchSnafu)?
124            .with_context(|| InvalidRecordBatchSnafu {
125                reason: format!("JSON2 column '{}' has no extension metadata", column.name),
126            })?;
127        // Source SSTs may use v1, but current region metadata is copied to the rewritten output.
128        // Since the target physical layout is always v2, v1 metadata would produce an
129        // inconsistent persisted field. The region opener normally upgraded this metadata already.
130        ensure!(
131            extension.metadata().is_version_2(),
132            InvalidRecordBatchSnafu {
133                reason: format!("JSON2 column '{}' is not layout v2", column.name),
134            }
135        );
136
137        let settings = extension.metadata().json_settings();
138        let hint_paths = settings
139            .type_hints()
140            .iter()
141            .map(|hint| hint.path.iter().map(String::as_str).collect::<Vec<_>>())
142            .collect::<HashSet<_>>();
143        let mut stats = HashMap::new();
144        for (schema, rows) in schemas {
145            let Some((_, field)) = schema.fields().find(&column.name) else {
146                continue;
147            };
148            collect_json2_path_stats(field, *rows, &hint_paths, &mut stats)?;
149        }
150
151        // A leaf absent from an input's explicit schema may have arbitrary values
152        // (including scalar ancestors) in its remainder. Only common explicit leaves
153        // are safe to promote without inspecting rows. Keep unsafe paths opaque for
154        // the entire output SST so existing readers never miss remainder values when
155        // projecting an explicit leaf.
156        for stat in stats.values_mut() {
157            stat.is_type_conflicted |= stat.sources != schemas.len();
158        }
159        let mut hints = settings.type_hints().to_vec();
160        hints.extend(select_dynamic_hints(settings, &hint_paths, &stats));
161        let target_layout = JsonSettings::try_new(hints, Some(0)).context(DataTypeMismatchSnafu)?;
162        plans.insert(
163            column.name.clone(),
164            Json2RewritePlan {
165                logical_settings: settings.clone(),
166                target_layout,
167            },
168        );
169    }
170    Ok(plans)
171}
172
173fn collect_json2_path_stats<'a>(
174    field: &'a Field,
175    rows: u64,
176    hint_paths: &HashSet<Vec<&str>>,
177    stats: &mut HashMap<Vec<&'a str>, Json2LeafPathStats>,
178) -> Result<()> {
179    let ArrowDataType::Struct(fields) = field.data_type() else {
180        return InvalidRecordBatchSnafu {
181            reason: format!("JSON2 column '{}' is not a struct", field.name()),
182        }
183        .fail();
184    };
185    let mut paths = Vec::new();
186    for field in fields {
187        if field.name() == JSON2_REMAINDER_FIELD_NAME {
188            continue;
189        }
190        collect_leaf_path_types(field, &mut Vec::new(), &mut paths)?;
191    }
192
193    for (path, data_type) in paths {
194        if hint_paths.contains(path.as_slice()) {
195            continue;
196        }
197        let Some(stat) = stats.get_mut(&path) else {
198            stats.insert(
199                path,
200                Json2LeafPathStats {
201                    rows,
202                    sources: 1,
203                    data_type,
204                    is_type_conflicted: false,
205                },
206            );
207            continue;
208        };
209        stat.sources += 1;
210        if stat.data_type != data_type {
211            stat.is_type_conflicted = true;
212        } else {
213            stat.rows += rows;
214        }
215    }
216    Ok(())
217}
218
219fn collect_leaf_path_types<'a>(
220    field: &'a Field,
221    path: &mut Vec<&'a str>,
222    paths: &mut Vec<(Vec<&'a str>, JsonNativeType)>,
223) -> Result<()> {
224    path.push(field.name());
225    if let ArrowDataType::Struct(fields) = field.data_type()
226        && !fields.is_empty()
227    {
228        for field in fields {
229            collect_leaf_path_types(field, path, paths)?;
230        }
231    } else {
232        let json_type =
233            JsonNativeType::try_from(field.data_type()).context(DataTypeMismatchSnafu)?;
234        paths.push((path.clone(), json_type));
235    }
236    path.pop();
237    Ok(())
238}
239
240fn select_dynamic_hints(
241    settings: &JsonSettings,
242    hint_paths: &HashSet<Vec<&str>>,
243    stats: &HashMap<Vec<&str>, Json2LeafPathStats>,
244) -> Vec<JsonTypeHint> {
245    let all_paths = stats
246        .keys()
247        .map(Vec::as_slice)
248        .chain(hint_paths.iter().map(Vec::as_slice))
249        .collect::<HashSet<_>>();
250    let has_ancestor_path =
251        |path: &[&str]| (1..path.len()).any(|len| all_paths.contains(&path[..len]));
252
253    let prefixes = all_paths
254        .iter()
255        .copied()
256        .flat_map(|path| (1..path.len()).map(|len| &path[..len]))
257        .collect::<HashSet<_>>();
258    let has_descendant_path = |path: &[&str]| prefixes.contains(path);
259
260    let mut candidates = stats
261        .iter()
262        .filter(|(path, stat)| {
263            !stat.is_type_conflicted
264                // TODO(LFC): Instead of "primitive only", consider retaining stable compound types
265                // that are safe to write to Parquet, as flush does. Or better, unite the two
266                // selection process.
267                && stat.data_type.is_primitive()
268                && !has_ancestor_path(path)
269                && !has_descendant_path(path)
270        })
271        .collect::<Vec<_>>();
272    candidates.sort_unstable_by(|(x_path, x), (y_path, y)| {
273        y.rows.cmp(&x.rows).then_with(|| x_path.cmp(y_path))
274    });
275    candidates
276        .into_iter()
277        .take(
278            settings
279                .max_auto_expanded_paths()
280                .unwrap_or(JSON2_DEFAULT_MAX_AUTO_EXPANDED_PATHS) as usize,
281        )
282        .map(|(path, stat)| JsonTypeHint {
283            path: path.iter().map(|x| (*x).to_owned()).collect(),
284            data_type: ConcreteDataType::from_arrow_type(&stat.data_type.as_arrow_type()),
285            inverted_index: false,
286            skipping_index: None,
287            fulltext_index: None,
288        })
289        .collect()
290}
291
292/// Replaces JSON2 physical field types according to the computed rewrite plans.
293pub(crate) fn rewrite_json2_schema(schema: &SchemaRef, plans: &Json2RewritePlans) -> SchemaRef {
294    if plans.is_empty() {
295        return schema.clone();
296    }
297    let fields = schema
298        .fields()
299        .iter()
300        .map(|field| {
301            let Some(plan) = plans.get(field.name()) else {
302                return field.clone();
303            };
304            let mut field = Field::clone(field);
305            field.set_data_type(json2_physical_data_type(&plan.target_layout));
306            field = field.with_extension_type(Json2ExtensionType::new(Arc::new(
307                JsonMetadata::new(plan.logical_settings.clone()),
308            )));
309            Arc::new(field)
310        })
311        .collect::<Vec<_>>();
312    Arc::new(Schema::new_with_metadata(fields, schema.metadata().clone()))
313}
314
315/// Rewrites JSON2 columns in `batch` according to the computed plans.
316pub(crate) fn rewrite_json2_batch(
317    batch: RecordBatch,
318    plans: &Json2RewritePlans,
319) -> Result<RecordBatch> {
320    if plans.is_empty() {
321        return Ok(batch);
322    }
323    let mut fields = Vec::with_capacity(batch.num_columns());
324    let mut columns = Vec::with_capacity(batch.num_columns());
325
326    for (field, array) in batch.schema_ref().fields().iter().zip(batch.columns()) {
327        let Some(plan) = plans.get(field.name()) else {
328            fields.push(field.clone());
329            columns.push(array.clone());
330            continue;
331        };
332
333        let array = JsonArray::from(array)
334            .rewrite_to_v2_with_type_hint_mismatch_policy(
335                field,
336                &plan.logical_settings,
337                &plan.target_layout,
338                TypeHintMismatchPolicy::CoerceOrNull,
339            )
340            .context(ConvertValueSnafu)?;
341        debug_assert_eq!(
342            &json2_physical_data_type(&plan.target_layout),
343            array.data_type()
344        );
345
346        let mut field = Field::clone(field);
347        field.set_data_type(array.data_type().clone());
348        field = field.with_extension_type(Json2ExtensionType::new(Arc::new(JsonMetadata::new(
349            plan.logical_settings.clone(),
350        ))));
351        fields.push(Arc::new(field));
352        columns.push(array);
353    }
354
355    let schema = Arc::new(Schema::new_with_metadata(
356        fields,
357        batch.schema_ref().metadata().clone(),
358    ));
359    RecordBatch::try_new(schema, columns).context(NewRecordBatchSnafu)
360}
361
362#[cfg(test)]
363mod tests {
364    use datatypes::extension::json::{
365        JSON2_REMAINDER_FIELD_NAME, Json2PhysicalLayout, JsonMetadata,
366    };
367    use datatypes::json::JsonTypeHint;
368    use datatypes::prelude::{ConcreteDataType, DataType};
369    use datatypes::schema::ColumnSchema;
370    use datatypes::types::json_type::{JsonNativeType, JsonObjectType};
371    use serde_json::json;
372
373    use super::*;
374
375    #[test]
376    fn test_select_dynamic_hints_rejects_type_and_prefix_conflicts()
377    -> Result<(), Box<dyn std::error::Error>> {
378        let settings = JsonSettings::try_new(
379            vec![JsonTypeHint {
380                path: vec!["hint".to_string()],
381                data_type: ConcreteDataType::string_datatype(),
382                inverted_index: false,
383                skipping_index: None,
384                fulltext_index: None,
385            }],
386            Some(2),
387        )?;
388        let stat = |rows, data_type, is_type_conflicted| Json2LeafPathStats {
389            rows,
390            sources: 1,
391            data_type,
392            is_type_conflicted,
393        };
394        let stats = HashMap::from([
395            (
396                vec!["hint", "nested"],
397                stat(10, JsonNativeType::String, false),
398            ),
399            (vec!["popular"], stat(9, JsonNativeType::String, false)),
400            (
401                vec!["popular", "nested"],
402                stat(8, JsonNativeType::u64(), false),
403            ),
404            (
405                vec!["type_conflicted"],
406                stat(7, JsonNativeType::String, true),
407            ),
408            (
409                vec!["array"],
410                stat(
411                    6,
412                    JsonNativeType::Array(Box::new(JsonNativeType::String)),
413                    false,
414                ),
415            ),
416            (vec!["variant"], stat(5, JsonNativeType::Variant, false)),
417            (vec!["tie_a"], stat(2, JsonNativeType::String, false)),
418            (vec!["tie_b"], stat(2, JsonNativeType::Bool, false)),
419        ]);
420
421        let hint_paths = settings
422            .type_hints()
423            .iter()
424            .map(|hint| hint.path.iter().map(String::as_str).collect::<Vec<_>>())
425            .collect::<HashSet<_>>();
426        let hints = select_dynamic_hints(&settings, &hint_paths, &stats);
427        assert_eq!(
428            vec![vec!["tie_a".to_string()], vec!["tie_b".to_string()]],
429            hints.into_iter().map(|x| x.path).collect::<Vec<_>>()
430        );
431        Ok(())
432    }
433
434    #[test]
435    fn test_rewrite_json2_v1_batch_to_target_layout() -> Result<(), Box<dyn std::error::Error>> {
436        let settings = JsonSettings::try_new(
437            vec![JsonTypeHint {
438                path: vec!["kind".to_string()],
439                data_type: ConcreteDataType::string_datatype(),
440                inverted_index: false,
441                skipping_index: None,
442                fulltext_index: None,
443            }],
444            Some(0),
445        )?;
446        let target = json2_physical_data_type(&settings);
447        let plans = HashMap::from([(
448            "j".to_string(),
449            Json2RewritePlan {
450                logical_settings: settings.clone(),
451                target_layout: settings,
452            },
453        )]);
454        let values = [
455            json!({"kind": "a", "extra": {"x": 1}}),
456            json!({"kind": "b", "extra": {"x": 2}}),
457        ];
458        let source_settings = JsonSettings::default();
459        let source_extension =
460            Json2ExtensionType::new(Arc::new(JsonMetadata::new_v1(source_settings.clone())));
461        let mut source_column = ColumnSchema::new(
462            "j",
463            ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())),
464            true,
465        );
466        source_column.with_extension_type(&source_extension);
467        let mut source_builder = source_column.data_type.create_mutable_vector(values.len());
468        for value in &values {
469            let value = source_settings.encode(value.clone())?;
470            source_builder.try_push_value_ref(&value.as_value_ref())?;
471        }
472        let source = source_builder.to_vector().to_arrow_array();
473        let field =
474            Field::new("j", source.data_type().clone(), true).with_extension_type(source_extension);
475        let batch = RecordBatch::try_new(Arc::new(Schema::new(vec![field])), vec![source])?;
476
477        let batch = rewrite_json2_batch(batch, &plans)?;
478        let field = batch.schema_ref().field(0);
479        assert!(Json2PhysicalLayout::try_from_root(field)?.is_version_2());
480        assert_eq!(&target, field.data_type());
481        let projected =
482            JsonArray::from(batch.column(0)).project_to_v2(field, &ArrowDataType::Binary)?;
483        let projected = JsonArray::from(&projected);
484        for (i, expected) in values.into_iter().enumerate() {
485            assert_eq!(expected, projected.try_get_value(i)?);
486        }
487        Ok(())
488    }
489
490    #[test]
491    fn test_rewrite_json2_v2_source_to_narrower_target() -> Result<(), Box<dyn std::error::Error>> {
492        let target_settings = JsonSettings::try_new(
493            vec![JsonTypeHint {
494                path: vec!["kind".to_string()],
495                data_type: ConcreteDataType::string_datatype(),
496                inverted_index: false,
497                skipping_index: None,
498                fulltext_index: None,
499            }],
500            Some(0),
501        )?;
502        let target_extension =
503            Json2ExtensionType::new(Arc::new(JsonMetadata::new(target_settings.clone())));
504        let target_type = json2_physical_data_type(&target_settings);
505        let plans = HashMap::from([(
506            "j".to_string(),
507            Json2RewritePlan {
508                logical_settings: target_settings.clone(),
509                target_layout: target_settings,
510            },
511        )]);
512
513        let source_settings = JsonSettings::try_new(
514            vec![
515                JsonTypeHint {
516                    path: vec!["kind".to_string()],
517                    data_type: ConcreteDataType::string_datatype(),
518                    inverted_index: false,
519                    skipping_index: None,
520                    fulltext_index: None,
521                },
522                JsonTypeHint {
523                    path: vec!["source_only".to_string()],
524                    data_type: ConcreteDataType::int64_datatype(),
525                    inverted_index: false,
526                    skipping_index: None,
527                    fulltext_index: None,
528                },
529            ],
530            None,
531        )?;
532        let source_extension =
533            Json2ExtensionType::new(Arc::new(JsonMetadata::new(source_settings.clone())));
534        let mut source_column = ColumnSchema::new(
535            "j",
536            ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())),
537            true,
538        );
539        source_column.with_extension_type(&source_extension);
540        let expected = json!({
541            "kind": "a",
542            "source_only": 7,
543            "dynamic": {"nested": true}
544        });
545        let mut builder = source_column.create_mutable_vector(1);
546        let value = source_settings.encode(expected.clone())?;
547        builder.try_push_value_ref(&value.as_value_ref())?;
548        let source = builder.to_vector().to_arrow_array();
549        let source_field =
550            Field::new("j", source.data_type().clone(), true).with_extension_type(source_extension);
551        let source =
552            JsonArray::from(&source).project_to_v2(&source_field, &ArrowDataType::Binary)?;
553        let field = Field::new("j", ArrowDataType::Binary, true)
554            .with_extension_type(target_extension.clone());
555        let batch = RecordBatch::try_new(Arc::new(Schema::new(vec![field])), vec![source])?;
556
557        let batch = rewrite_json2_batch(batch, &plans)?;
558        let field = batch.schema_ref().field(0);
559        assert!(Json2PhysicalLayout::try_from_root(field)?.is_version_2());
560        assert_eq!(&target_type, field.data_type());
561        let ArrowDataType::Struct(fields) = field.data_type() else {
562            unreachable!()
563        };
564        assert_eq!(
565            vec![JSON2_REMAINDER_FIELD_NAME, "kind"],
566            fields.iter().map(|x| x.name().as_str()).collect::<Vec<_>>()
567        );
568
569        let projected =
570            JsonArray::from(batch.column(0)).project_to_v2(field, &ArrowDataType::Binary)?;
571        assert_eq!(expected, JsonArray::from(&projected).try_get_value(0)?);
572
573        let first_schema = batch.schema();
574        let field =
575            Field::new("j", ArrowDataType::Binary, true).with_extension_type(target_extension);
576        let batch = RecordBatch::try_new(Arc::new(Schema::new(vec![field])), vec![projected])?;
577        let batch = rewrite_json2_batch(batch, &plans)?;
578        assert_eq!(first_schema, batch.schema());
579
580        let field = batch.schema_ref().field(0);
581        let projected =
582            JsonArray::from(batch.column(0)).project_to_v2(field, &ArrowDataType::Binary)?;
583        assert_eq!(expected, JsonArray::from(&projected).try_get_value(0)?);
584        Ok(())
585    }
586
587    #[test]
588    fn test_rewrite_json2_v2_source_to_wider_target() -> Result<(), Box<dyn std::error::Error>> {
589        let logical_settings = JsonSettings::try_new(
590            vec![JsonTypeHint {
591                path: vec!["kind".to_string()],
592                data_type: ConcreteDataType::string_datatype(),
593                inverted_index: false,
594                skipping_index: None,
595                fulltext_index: None,
596            }],
597            Some(0),
598        )?;
599        let target_layout = JsonSettings::try_new(
600            vec![
601                JsonTypeHint {
602                    path: vec!["kind".to_string()],
603                    data_type: ConcreteDataType::string_datatype(),
604                    inverted_index: false,
605                    skipping_index: None,
606                    fulltext_index: None,
607                },
608                JsonTypeHint {
609                    path: vec!["promoted".to_string()],
610                    data_type: ConcreteDataType::int64_datatype(),
611                    inverted_index: false,
612                    skipping_index: None,
613                    fulltext_index: None,
614                },
615            ],
616            Some(0),
617        )?;
618        let target_type = json2_physical_data_type(&target_layout);
619        let plans = HashMap::from([(
620            "j".to_string(),
621            Json2RewritePlan {
622                logical_settings: logical_settings.clone(),
623                target_layout,
624            },
625        )]);
626
627        let extension =
628            Json2ExtensionType::new(Arc::new(JsonMetadata::new(logical_settings.clone())));
629        let mut column = ColumnSchema::new(
630            "j",
631            ConcreteDataType::json2(JsonNativeType::Object(JsonObjectType::new())),
632            true,
633        );
634        column.with_extension_type(&extension);
635        let expected = json!({
636            "kind": "a",
637            "promoted": 7,
638            "dynamic": {"nested": true}
639        });
640        let mut builder = column.create_mutable_vector(1);
641        let value = logical_settings.encode(expected.clone())?;
642        builder.try_push_value_ref(&value.as_value_ref())?;
643        let source = builder.to_vector().to_arrow_array();
644        assert_eq!(
645            &json2_physical_data_type(&logical_settings),
646            source.data_type()
647        );
648        let field =
649            Field::new("j", source.data_type().clone(), true).with_extension_type(extension);
650        let batch = RecordBatch::try_new(Arc::new(Schema::new(vec![field])), vec![source])?;
651
652        let batch = rewrite_json2_batch(batch, &plans)?;
653        let field = batch.schema_ref().field(0);
654        assert_eq!(&target_type, field.data_type());
655        let ArrowDataType::Struct(fields) = field.data_type() else {
656            unreachable!()
657        };
658        assert_eq!(
659            vec![JSON2_REMAINDER_FIELD_NAME, "kind", "promoted"],
660            fields.iter().map(|x| x.name().as_str()).collect::<Vec<_>>()
661        );
662
663        let array = batch
664            .column(0)
665            .as_any()
666            .downcast_ref::<datatypes::arrow::array::StructArray>()
667            .unwrap();
668        let promoted = array
669            .column_by_name("promoted")
670            .unwrap()
671            .as_any()
672            .downcast_ref::<datatypes::arrow::array::Int64Array>()
673            .unwrap();
674        assert_eq!(7, promoted.value(0));
675
676        let projected =
677            JsonArray::from(batch.column(0)).project_to_v2(field, &ArrowDataType::Binary)?;
678        assert_eq!(expected, JsonArray::from(&projected).try_get_value(0)?);
679        Ok(())
680    }
681}
mito2/compaction/json2.rs

mito2/compaction/
json2.rs
