fix(fuse): validate inline variant bytes

Author: dantengsky

src/query/storages/fuse/src/io/write/virtual_column_builder.rs

@@ -793,31 +793,29 @@ impl VirtualColumnBuilder {
 
         for row in 0..num_rows {
             let mut decoded_jsonb: Option<Vec<u8>> = None;
+            let mut validated_jsonb = false;
             for (idx, builder) in builders.iter_mut().enumerate() {
                 let entry = block.get_by_offset(source_offsets[idx]);
                 let value = unsafe { entry.index_unchecked(row) };
                 let ScalarRef::Variant(jsonb_bytes) = value else {
                     builder.push(ScalarRef::Null);
                     continue;
                 };
-                let mut raw_jsonb = RawJsonb::new(jsonb_bytes);
+                if !validated_jsonb {
+                    validated_jsonb = true;
+                    if jsonb::from_slice(jsonb_bytes).is_err() {
+                        if let Ok(jsonb) = parquet_variant_bytes_to_jsonb(jsonb_bytes) {
+                            decoded_jsonb = Some(jsonb);
+                        }
+                    }
+                }
+                let jsonb_bytes = decoded_jsonb.as_deref().unwrap_or(jsonb_bytes);
+                let raw_jsonb = RawJsonb::new(jsonb_bytes);
                 let path = &borrowed_paths[idx];
                 let raw_value = match raw_jsonb.get_by_keypath(path.paths.iter()) {
                     Ok(Some(value)) => Some(value),
                     Ok(None) => None,
-                    Err(_) => {
-                        if decoded_jsonb.is_none() {
-                            if let Ok(jsonb) = parquet_variant_bytes_to_jsonb(jsonb_bytes) {
-                                decoded_jsonb = Some(jsonb);
-                            }
-                        }
-                        if let Some(ref jsonb) = decoded_jsonb {
-                            raw_jsonb = RawJsonb::new(jsonb);
-                            raw_jsonb.get_by_keypath(path.paths.iter()).ok().flatten()
-                        } else {
-                            None
-                        }
-                    }
+                    Err(_) => None,
                 };
                 let Some(raw_value) = raw_value else {
                     builder.push(ScalarRef::Null);

tests/sqllogictests/suites/ee/01_ee_system/01_0002_virtual_column.test

@@ -143,29 +143,29 @@ test_virtual_column t2 val 3000000002 ['c'] String
 query II
 select row_count, virtual_column_size from fuse_block('test_virtual_column', 't2')
 ----
-3 1032
+3 809
 
 query III
 select block_count, row_count, virtual_column_size from fuse_segment('test_virtual_column', 't2');
 ----
-1 3 1032
+1 3 809
 
 query III
 select block_count, row_count, virtual_column_size from fuse_snapshot('test_virtual_column', 't2');
 ----
-1 3 1032
+1 3 809
 
 query IITTIII
 select virtual_block_size, row_count, column_name, column_type, column_id, block_offset, bytes_compressed from fuse_virtual_column('test_virtual_column', 't2')
 ----
-1032 3 val['a'] UInt64 NULL 0 4 40
-1032 3 val['b'] UInt64 NULL 1 44 40
-1032 3 val['c'] String NULL 2 84 40
+809 3 val['a'] UInt64 NULL 3000000000 4 40
+809 3 val['b'] UInt64 NULL 3000000001 44 40
+809 3 val['c'] String NULL 3000000002 84 40
 
 query IIIIII
 select block_count, row_count, bytes_uncompressed, bytes_compressed, index_size, virtual_block_count from fuse_segment('test_virtual_column', 't2')
 ----
-1 3 134 734 1470 1
+1 3 134 734 1247 1
 
 statement ok
 insert into t2 values(4, '{"a":44,"b":4,"c":"value"}'), (5, '{"a":55,"b":5,"c":"bend"}'), (6, '6')
@@ -343,6 +343,8 @@ test_virtual_column tweets data['create'] 10 10 10 0 1/08 6/07 16 (empty)
 test_virtual_column tweets data['id'] 10 10 10 0 1 10 8 (empty)
 test_virtual_column tweets data['likes'] 10 10 2 0 10 25 1 (empty)
 test_virtual_column tweets data['replies'] 10 10 7 3 0 9 8 (empty)
+test_virtual_column tweets data['tags'][0] 10 10 2 0 good new 3 (empty)
+test_virtual_column tweets data['tags'][1] 10 10 2 0 interesting popular 3 (empty)
 test_virtual_column tweets data['text'] 10 10 7 0 a z 16 (empty)
 test_virtual_column tweets data['user']['id'] 10 10 6 0 1 7 8 (empty)
 test_virtual_column tweets id 10 10 10 0 1 10 4 (empty)
@@ -360,6 +362,8 @@ data['replies'] 7 3 8 (empty)
 data['text'] 8 0 16 (empty)
 data['user']['id'] 6 0 8 (empty)
 data['likes'] 2 0 1 (empty)
+data['tags'][0] 2 0 3 (empty)
+data['tags'][1] 2 0 3 (empty)
 
 query TTTIIIITTIT
 SHOW STATISTICS FROM TABLE test_virtual_column.tweets;
@@ -369,6 +373,8 @@ test_virtual_column tweets data['create'] 10 10 10 0 1/08 6/07 16 (empty)
 test_virtual_column tweets data['id'] 10 10 10 0 1 10 8 (empty)
 test_virtual_column tweets data['likes'] 10 10 2 0 10 25 1 (empty)
 test_virtual_column tweets data['replies'] 10 10 7 3 0 10 8 (empty)
+test_virtual_column tweets data['tags'][0] 10 10 2 0 good new 3 (empty)
+test_virtual_column tweets data['tags'][1] 10 10 2 0 interesting popular 3 (empty)
 test_virtual_column tweets data['text'] 10 10 8 0 a z 16 (empty)
 test_virtual_column tweets data['user']['id'] 10 10 6 0 1 7 8 (empty)
 test_virtual_column tweets id 10 10 10 0 1 10 4 (empty)
@@ -493,7 +499,7 @@ select id, data['id'], data['create'], data['text'], data['user']['id'], data['r
 query I
 SELECT count(*) FROM fuse_virtual_column('test_virtual_column', 'tweets');
 ----
-27
+26
 
 statement ok
 set enable_experimental_virtual_column = 0;
@@ -504,7 +510,7 @@ INSERT INTO tweets FROM (SELECT * FROM tweets);
 query I
 SELECT count(*) FROM fuse_virtual_column('test_virtual_column', 'tweets');
 ----
-27
+26
 
 statement error
 REFRESH VIRTUAL COLUMN FOR tweets;
@@ -522,7 +528,7 @@ optimize table test_virtual_column.tweets compact
 query I
 SELECT count(*) FROM fuse_virtual_column('test_virtual_column', 'tweets');
 ----
-8
+6
 
 query IITTIITFIT
 select id, data['id'], data['create'], data['text'], data['user']['id'], data['replies'], data['geo'], data['geo']['lat'], data['likes'], data['tags'] from tweets order by id;
@@ -705,8 +711,11 @@ test_virtual_column test_stream data1 3000000000 ['address']['city'] String
 test_virtual_column test_stream data1 3000000001 ['address']['district'] String
 test_virtual_column test_stream data1 3000000002 ['age'] UInt64
 test_virtual_column test_stream data1 3000000003 ['name'] String
-test_virtual_column test_stream data2 3000000004 ['hobbies'] Jsonb
-test_virtual_column test_stream data2 3000000005 ['scores'] Jsonb
+test_virtual_column test_stream data2 3000000004 ['hobbies'][0] String
+test_virtual_column test_stream data2 3000000005 ['hobbies'][1] String
+test_virtual_column test_stream data2 3000000006 ['scores'][0] UInt64
+test_virtual_column test_stream data2 3000000007 ['scores'][1] UInt64
+test_virtual_column test_stream data2 3000000008 ['scores'][2] UInt64
 
 query ITTTTT
 SELECT