fix(query): fix cast nested types cause panic (#17622)

* fix(query): fix cast nested types cause panic

* fix

* fix typo

* fix

* fix

* fix

Author: b41sh

src/common/native/src/nested.rs

@@ -171,18 +171,18 @@ fn to_nested_recursive(
         Column::Array(inner) => {
             parents.push(Nested::LargeList(ListNested {
                 is_nullable: nullable,
-                offsets: inner.offsets.clone(),
+                offsets: inner.underlying_offsets(),
                 validity,
             }));
-            to_nested_recursive(&inner.values, nested, parents)?;
+            to_nested_recursive(&inner.underlying_column(), nested, parents)?;
         }
         Column::Map(inner) => {
             parents.push(Nested::LargeList(ListNested {
                 is_nullable: nullable,
-                offsets: inner.offsets.clone(),
+                offsets: inner.underlying_offsets(),
                 validity,
             }));
-            to_nested_recursive(&inner.values, nested, parents)?;
+            to_nested_recursive(&inner.underlying_column(), nested, parents)?;
         }
         _ => {
             parents.push(Nested::Primitive(column.len(), nullable, validity));
@@ -206,10 +206,10 @@ fn to_leaves_recursive(column: &Column, leaves: &mut Vec<Column>) {
             cs.iter().for_each(|a| to_leaves_recursive(a, leaves));
         }
         Column::Array(col) => {
-            to_leaves_recursive(&col.values, leaves);
+            to_leaves_recursive(&col.underlying_column(), leaves);
         }
         Column::Map(col) => {
-            to_leaves_recursive(&col.values, leaves);
+            to_leaves_recursive(&col.underlying_column(), leaves);
         }
         // Handle nullable columns by recursing into their inner value
         Column::Nullable(inner) => to_leaves_recursive(&inner.column, leaves),
@@ -243,7 +243,7 @@ impl InitNested {
 pub fn create_list(data_type: TableDataType, nested: &mut NestedState, values: Column) -> Column {
     let n = nested.pop().unwrap();
     let (offsets, validity) = n.inner();
-    let col = Column::Array(Box::new(ArrayColumn::<AnyType> { values, offsets }));
+    let col = Column::Array(Box::new(ArrayColumn::<AnyType>::new(values, offsets)));
 
     if data_type.is_nullable() {
         col.wrap_nullable(validity.clone())
@@ -256,7 +256,7 @@ pub fn create_list(data_type: TableDataType, nested: &mut NestedState, values: C
 pub fn create_map(data_type: TableDataType, nested: &mut NestedState, values: Column) -> Column {
     let n = nested.pop().unwrap();
     let (offsets, validity) = n.inner();
-    let col = Column::Map(Box::new(ArrayColumn::<AnyType> { values, offsets }));
+    let col = Column::Map(Box::new(ArrayColumn::<AnyType>::new(values, offsets)));
     if data_type.is_nullable() {
         col.wrap_nullable(validity.clone())
     } else {

src/query/expression/src/converts/arrow/from.rs

@@ -315,7 +315,7 @@ impl Column {
                 let values = Column::from_arrow_rs(array.values().clone(), inner.as_ref())?;
                 let offsets: Buffer<u64> = array.offsets().inner().inner().clone().into();
 
-                let inner_col = ArrayColumn { values, offsets };
+                let inner_col = ArrayColumn::new(values, offsets);
                 Column::Array(Box::new(inner_col))
             }
             DataType::Map(inner) => {
@@ -333,7 +333,7 @@ impl Column {
                 let offsets: Buffer<i32> = array.offsets().inner().inner().clone().into();
                 let offsets = offsets.into_iter().map(|x| x as u64).collect();
 
-                let inner_col = ArrayColumn { values, offsets };
+                let inner_col = ArrayColumn::new(values, offsets);
                 Column::Map(Box::new(inner_col))
             }
             DataType::Tuple(ts) => {

src/query/expression/src/converts/arrow/to.rs

@@ -303,10 +303,10 @@ impl From<&Column> for ArrayData {
             Column::Date(col) => buffer_to_array_data((col.clone(), arrow_type)),
             Column::Interval(col) => buffer_to_array_data((col.clone(), arrow_type)),
             Column::Array(col) => {
-                let child_data = ArrayData::from(&col.values);
+                let child_data = ArrayData::from(&col.underlying_column());
                 let builder = ArrayDataBuilder::new(arrow_type)
                     .len(value.len())
-                    .buffers(vec![col.offsets.clone().into()])
+                    .buffers(vec![col.underlying_offsets().into()])
                     .child_data(vec![child_data]);
 
                 unsafe { builder.build_unchecked() }
@@ -318,8 +318,12 @@ impl From<&Column> for ArrayData {
                 unsafe { builder.nulls(Some(nulls)).build_unchecked() }
             }
             Column::Map(col) => {
-                let child_data = ArrayData::from(&col.values);
-                let offsets: Vec<i32> = col.offsets.iter().map(|x| *x as i32).collect();
+                let child_data = ArrayData::from(&col.underlying_column());
+                let offsets: Vec<i32> = col
+                    .underlying_offsets()
+                    .into_iter()
+                    .map(|x| x as i32)
+                    .collect();
                 let builder = ArrayDataBuilder::new(arrow_type)
                     .len(value.len())
                     .buffers(vec![offsets.into()])

src/query/expression/src/converts/meta/bincode.rs

@@ -144,14 +144,14 @@ impl From<LegacyColumn> for Column {
             LegacyColumn::Timestamp(buf) => Column::Timestamp(buf),
             LegacyColumn::Date(buf) => Column::Date(buf),
             LegacyColumn::Interval(buf) => Column::Interval(buf),
-            LegacyColumn::Array(arr_col) => Column::Array(Box::new(ArrayColumn::<AnyType> {
-                values: arr_col.values.into(),
-                offsets: arr_col.offsets,
-            })),
-            LegacyColumn::Map(map_col) => Column::Map(Box::new(ArrayColumn::<AnyType> {
-                values: map_col.values.into(),
-                offsets: map_col.offsets,
-            })),
+            LegacyColumn::Array(arr_col) => Column::Array(Box::new(ArrayColumn::<AnyType>::new(
+                arr_col.values.into(),
+                arr_col.offsets,
+            ))),
+            LegacyColumn::Map(map_col) => Column::Map(Box::new(ArrayColumn::<AnyType>::new(
+                map_col.values.into(),
+                map_col.offsets,
+            ))),
             LegacyColumn::Bitmap(str_col) => Column::Bitmap(BinaryColumn::from(str_col)),
             LegacyColumn::Nullable(nullable_col) => {
                 Column::Nullable(Box::new(NullableColumn::<AnyType> {
@@ -207,12 +207,12 @@ impl From<Column> for LegacyColumn {
             Column::Date(buf) => LegacyColumn::Date(buf),
             Column::Interval(buf) => LegacyColumn::Interval(buf),
             Column::Array(arr_col) => LegacyColumn::Array(Box::new(LegacyArrayColumn {
-                values: arr_col.values.into(),
-                offsets: arr_col.offsets,
+                values: arr_col.underlying_column().into(),
+                offsets: arr_col.underlying_offsets(),
             })),
             Column::Map(map_col) => LegacyColumn::Map(Box::new(LegacyArrayColumn {
-                values: map_col.values.into(),
-                offsets: map_col.offsets,
+                values: map_col.underlying_column().into(),
+                offsets: map_col.underlying_offsets(),
             })),
             Column::Bitmap(str_col) => LegacyColumn::Bitmap(LegacyBinaryColumn::from(str_col)),
             Column::Nullable(nullable_col) => {