Merge branch 'main' into dev-unescape-string

Author: mergify[bot]

Cargo.lock

@@ -34,12 +34,13 @@ simd = ["arrow/simd"]
 # Workspace dependencies
 
 # Crates.io dependencies
-arrow = { package = "arrow2", git = "https://github.com/jorgecarleitao/arrow2", rev = "95e117d", default-features = false, features = [
+arrow = { package = "arrow2", git = "https://github.com/jorgecarleitao/arrow2", rev = "1da33ac", default-features = false, features = [
     "io_parquet",
     "io_parquet_compression",
 ] }
+
 arrow-format = { version = "0.8.0", features = ["flight-data", "flight-service", "ipc"] }
 futures = "0.3.24"
-parquet2 = { version = "0.16.3", default_features = false }
+parquet2 = { version = "0.17.0", default_features = false }
 
 [dev-dependencies]

src/common/arrow/Cargo.toml

@@ -40,6 +40,7 @@ pub fn serialize_data_blocks_with_compression(
         write_statistics: false,
         compression,
         version: Version::V2,
+        data_pagesize_limit: None,
     };
     let batches = blocks
         .into_iter()
@@ -87,16 +88,12 @@ pub fn serialize_data_blocks(
     serialize_data_blocks_with_compression(blocks, schema, buf, CompressionOptions::Lz4Raw)
 }
 
-fn col_encoding(_data_type: &ArrowDataType) -> Encoding {
-    // Although encoding does work, parquet2 has not implemented decoding of DeltaLengthByteArray yet, we fallback to Plain
-    // From parquet2: Decoding "DeltaLengthByteArray"-encoded required V2 pages is not yet implemented for Binary.
-    //
-    // match data_type {
-    //    ArrowDataType::Binary
-    //    | ArrowDataType::LargeBinary
-    //    | ArrowDataType::Utf8
-    //    | ArrowDataType::LargeUtf8 => Encoding::DeltaLengthByteArray,
-    //    _ => Encoding::Plain,
-    //}
-    Encoding::Plain
+fn col_encoding(data_type: &ArrowDataType) -> Encoding {
+    match data_type {
+        ArrowDataType::Binary
+        | ArrowDataType::LargeBinary
+        | ArrowDataType::Utf8
+        | ArrowDataType::LargeUtf8 => Encoding::DeltaLengthByteArray,
+        _ => Encoding::Plain,
+    }
 }

src/query/datablocks/src/serialize.rs

@@ -37,6 +37,7 @@ pub struct InputFormatTSV {}
 impl InputFormatTSV {
     #[allow(clippy::too_many_arguments)]
     fn read_row(
+        field_delimiter: u8,
         field_decoder: &FieldDecoderTSV,
         buf: &[u8],
         deserializers: &mut Vec<common_datavalues::TypeDeserializerImpl>,
@@ -53,7 +54,7 @@ impl InputFormatTSV {
         let mut err_msg = None;
         let buf_len = buf.len();
         while pos <= buf_len {
-            if pos == buf_len || buf[pos] == b'\t' {
+            if pos == buf_len || buf[pos] == field_delimiter {
                 let col_data = &buf[field_start..pos];
                 if col_data.is_empty() {
                     deserializers[column_index].de_default();
@@ -156,6 +157,7 @@ impl InputFormatTextBase for InputFormatTSV {
         for (i, end) in batch.row_ends.iter().enumerate() {
             let buf = &batch.data[start..*end]; // include \n
             Self::read_row(
+                builder.ctx.field_delimiter,
                 field_decoder,
                 buf,
                 columns,

src/query/pipeline/sources/src/processors/sources/input_formats/impls/input_format_tsv.rs

@@ -242,15 +242,14 @@ impl MySQLFederated {
                 None,
             ),
             ("(?i)^(/\\* ApplicationName=(.*)SHOW VARIABLES(.*))", None),
-            // pt-toolkit
-            ("(?i)^(/\\*!40101 SET(.*) \\*/)$", None),
             // mysqldump 5.7.16
             ("(?i)^(/\\*!40100 SET(.*) \\*/)$", None),
             ("(?i)^(/\\*!40103 SET(.*) \\*/)$", None),
             ("(?i)^(/\\*!40111 SET(.*) \\*/)$", None),
             ("(?i)^(/\\*!40101 SET(.*) \\*/)$", None),
             ("(?i)^(/\\*!40014 SET(.*) \\*/)$", None),
             ("(?i)^(/\\*!40000 SET(.*) \\*/)$", None),
+            ("(?i)^(/\\*!40000 ALTER(.*) \\*/)$", None),
         ];
 
         FederatedHelper::block_match_rule(query, rules)

src/query/service/src/servers/mysql/mysql_federated.rs

@@ -430,6 +430,9 @@ impl BlockReader {
                 Err(ErrorCode::StorageOther(err_msg))
             }
             Compression::Lz4Raw => Ok(ParquetCompression::Lz4Raw),
+            Compression::Snappy => Ok(ParquetCompression::Snappy),
+            Compression::Zstd => Ok(ParquetCompression::Zstd),
+            Compression::Gzip => Ok(ParquetCompression::Gzip),
         }
     }
 

src/query/storages/fuse/fuse/src/io/read/block_reader.rs

@@ -85,6 +85,9 @@ where Self: Sized
 pub enum Compression {
     Lz4,
     Lz4Raw,
+    Snappy,
+    Zstd,
+    Gzip,
 }
 
 impl Compression {