fix: should use is_nested while populating in-memory cache (databen…

…dlabs#14503)

* fix: should use `is_nested` during populating in-mem cache

* revert modification of config `databend-query-node-1.toml`

src/query/service/tests/it/storages/fuse/io.rs

@@ -12,10 +12,14 @@
 //  See the License for the specific language governing permissions and
 //  limitations under the License.
 
+use databend_common_config::InnerConfig;
 use databend_common_exception::Result;
+use databend_common_expression::DataBlock;
 use databend_query::storages::fuse::io::TableMetaLocationGenerator;
+use databend_query::test_kits::TestFixture;
 use databend_storages_common_table_meta::meta::TableSnapshot;
 use databend_storages_common_table_meta::meta::Versioned;
+use futures_util::TryStreamExt;
 use uuid::Uuid;
 
 #[test]
@@ -31,3 +35,48 @@ fn test_meta_locations() -> Result<()> {
     assert!(snapshot_loc.starts_with(test_prefix));
     Ok(())
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn test_array_cache_of_nested_column_iusse_14502() -> Result<()> {
+    // https://github.com/datafuselabs/databend/issues/14502
+    // ~~~
+    //  create table t1(c tuple(c1 int not null, c2 int null) null);
+    //  insert into t1 values((1,null));
+    //  select c.1 from t1;
+    //  select c from t1;
+    // ~~~
+
+    let mut config = InnerConfig::default();
+    // memory cache is not enabled by default, let's enable it
+    config.cache.table_data_deserialized_data_bytes = 1024 * 1024 * 10;
+    let fixture = TestFixture::setup_with_config(&config).await?;
+
+    fixture.create_default_database().await?;
+    let db = fixture.default_db_name();
+
+    let sql_create = format!(
+        "create table {db}.t1(c tuple(c1 int not null, c2 int null) null) storage_format = Parquet"
+    );
+    let insert = format!("insert into {db}.t1 values((1,null))");
+    let q1 = format!("select c.1 from {db}.t1");
+    let q2 = format!("select c from {db}.t1");
+
+    let stmts = vec![sql_create, insert];
+
+    for x in &stmts {
+        fixture.execute_command(x).await?;
+    }
+
+    let queries = vec![q1, q2];
+
+    for x in &queries {
+        let res = fixture
+            .execute_query(x)
+            .await?
+            .try_collect::<Vec<DataBlock>>()
+            .await;
+        assert!(res.is_ok());
+    }
+
+    Ok(())
+}

src/query/storages/fuse/src/io/read/block/block_reader_parquet_deserialize.rs

@@ -247,7 +247,7 @@ impl BlockReader {
         let uncompressed_buffer = deserialization_context.uncompressed_buffer;
         // column passed in may be a compound field (with sub leaves),
         // or a leaf column of compound field
-        let is_nested = column.has_children();
+        let is_nested = column.is_nested;
         let estimated_cap = indices.len();
         let mut field_column_metas = Vec::with_capacity(estimated_cap);
         let mut field_column_data = Vec::with_capacity(estimated_cap);