[fix](lzo) fix lzo decompression failed

be/src/exec/lzo_decompressor.cpp

@@ -80,8 +80,6 @@ Status LzopDecompressor::decompress(uint8_t* input, size_t input_len, size_t* in
         }
     }
 
-    // LOG(INFO) << "after load header: " << *input_bytes_read;
-
     // read compressed block
     // compressed-block ::=
     //   <uncompressed-size>
@@ -144,7 +142,7 @@ Status LzopDecompressor::decompress(uint8_t* input, size_t input_len, size_t* in
             return Status::OK();
         }
 
-        ptr = get_uint32(ptr, &out_checksum);
+        ptr = get_uint32(ptr, &in_checksum);
         left_input_len -= sizeof(uint32_t);
     } else {
         // If the compressed data size is equal to the uncompressed data size, then
@@ -185,25 +183,15 @@ Status LzopDecompressor::decompress(uint8_t* input, size_t input_len, size_t* in
         ptr += compressed_size;
     }
 
-    // 7. peek next block's uncompressed size
-    uint32_t next_uncompressed_size;
-    get_uint32(ptr, &next_uncompressed_size);
-    if (next_uncompressed_size == 0) {
-        // 0 means current block is the last block.
-        // consume this uncompressed_size to finish reading.
-        ptr += sizeof(uint32_t);
-    }
-
-    // 8. done
+    // 7. done
     *stream_end = true;
     *decompressed_len = uncompressed_size;
     *input_bytes_read += ptr - block_start;
 
     VLOG_DEBUG << "finished decompress lzo block."
                << " compressed_size: " << compressed_size
                << " decompressed_len: " << *decompressed_len
-               << " input_bytes_read: " << *input_bytes_read
-               << " next_uncompressed_size: " << next_uncompressed_size;
+               << " input_bytes_read: " << *input_bytes_read;
 
     return Status::OK();
 }

be/src/vec/exec/format/file_reader/new_plain_text_line_reader.cpp

@@ -425,12 +425,12 @@ Status NewPlainTextLineReader::read_line(const uint8_t** ptr, size_t* size, bool
                         _output_buf_size - _output_buf_limit,               /* output_max_len */
                         &decompressed_len, &stream_end, &_more_input_bytes, &_more_output_bytes));
 
-                // LOG(INFO) << "after decompress:"
-                //           << " stream_end: " << stream_end
-                //           << " input_read_bytes: " << input_read_bytes
-                //           << " decompressed_len: " << decompressed_len
-                //           << " more_input_bytes: " << _more_input_bytes
-                //           << " more_output_bytes: " << _more_output_bytes;
+                VLOG_DEBUG << "after decompress:"
+                           << " stream_end: " << stream_end
+                           << " input_read_bytes: " << input_read_bytes
+                           << " decompressed_len: " << decompressed_len
+                           << " more_input_bytes: " << _more_input_bytes
+                           << " more_output_bytes: " << _more_output_bytes;
 
                 // update pos and limit
                 _input_buf_pos += input_read_bytes;

regression-test/data/external_table_p0/tvf/test_local_tvf_lzo.out

@@ -0,0 +1,29 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !test_1 --
+This is the first block of test data for LZO compression!11111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111111
+This is the third block with more test data for LZO compression!22222222222222222222222222222222222222222222222222222222222222222222222222222222222222222
+
+-- !test_2 --
+This is the first block of test data for LZO compression!
+This is the third block with more test data for LZO compression!
+
+-- !test_3 --
+This is the first block of test data for LZO compression!
+This is the third block with more test data for LZO compression!
+
+-- !test_4 --
+This is the first block of test data for LZO compression!
+This is the third block with more test data for LZO compression!
+
+-- !test_5 --
+2
+
+-- !test_6 --
+2
+
+-- !test_7 --
+2
+
+-- !test_8 --
+2
+

regression-test/suites/external_table_p0/tvf/test_local_tvf_lzo.groovy

@@ -0,0 +1,58 @@
+import org.junit.Assert
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// This suit test the `backends` tvf
+suite("test_local_tvf_lzo", "p0,external,external_docker") {
+    List<List<Object>> backends =  sql """ show backends """
+    def dataFilePath = context.config.dataPath + "/external_table_p0/tvf/lzo"
+
+    assertTrue(backends.size() > 0)
+
+    def be_id = backends[0][0]
+    // cluster mode need to make sure all be has this data
+    def outFilePath="/"
+    def transFile01="${dataFilePath}/test_compress.lzo"
+    def transFile02="${dataFilePath}/test_no_compress_with_empty_block_begin.lzo"
+    def transFile03="${dataFilePath}/test_no_compress_with_empty_block_end.lzo"
+    def transFile04="${dataFilePath}/test_no_compress_with_empty_block_middle.lzo"
+
+    for (List<Object> backend : backends) {
+        def be_host = backend[1]
+        scpFiles ("root", be_host, transFile01, outFilePath, false);
+        scpFiles ("root", be_host, transFile02, outFilePath, false);
+        scpFiles ("root", be_host, transFile03, outFilePath, false);
+        scpFiles ("root", be_host, transFile04, outFilePath, false);
+    }
+
+    def file1 = outFilePath + "test_compress.lzo";
+    def file2 = outFilePath + "test_no_compress_with_empty_block_begin.lzo";
+    def file3 = outFilePath + "test_no_compress_with_empty_block_end.lzo";
+    def file4 = outFilePath + "test_no_compress_with_empty_block_middle.lzo";
+
+    order_qt_test_1 """ select * from local( "file_path" = "${file1}", "backend_id" = "${be_id}", "format" = "csv");"""
+    order_qt_test_2 """ select * from local( "file_path" = "${file2}", "backend_id" = "${be_id}", "format" = "csv");"""
+    order_qt_test_3 """ select * from local( "file_path" = "${file3}", "backend_id" = "${be_id}", "format" = "csv");"""
+    order_qt_test_4 """ select * from local( "file_path" = "${file4}", "backend_id" = "${be_id}", "format" = "csv");"""
+
+    qt_test_5 """ select count(*) from local( "file_path" = "${file1}", "backend_id" = "${be_id}", "format" = "csv");"""
+    qt_test_6 """ select count(*) from local( "file_path" = "${file2}", "backend_id" = "${be_id}", "format" = "csv");"""
+    qt_test_7 """ select count(*) from local( "file_path" = "${file3}", "backend_id" = "${be_id}", "format" = "csv");"""
+    qt_test_8 """ select count(*) from local( "file_path" = "${file4}", "backend_id" = "${be_id}", "format" = "csv");"""
+
+}

tools/lzo/README.md

@@ -0,0 +1,37 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# LZO Writer
+
+This is simple LZO writer, you can use it to generate lzo compressed file.
+
+See `main()` for details
+
+## Compilation
+
+```
+sh build.sh
+```
+
+## Generate file
+
+```
+./lzo_writer file_name.lzo
+```
+

tools/lzo/build.sh

@@ -0,0 +1,20 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")" &>/dev/null && pwd)"
+export DORIS_THIRDPARTY=${ROOT}/../../thirdparty/
+g++ -o lzo_writer lzo_writer.cpp -I. -Isrc -I${DORIS_THIRDPARTY}/installed/include -L${DORIS_THIRDPARTY}/installed/lib -llzo2 -std=c++17