[fix](load)fix ingestion load error case cause be core. (#55906)

Related PR: #45937
branch-3.1: #55500

Problem Summary:
Fix the error case on ingestion load and the core in parquet reader.

```
==8898==ERROR: AddressSanitizer: heap-buffer-overflow on address 0x62f0020603fc at pc 0x55f634e64ded bp 0x7fba0d03c410 sp 0x7fba0d03bbd8 READ of size 4 at 0x62f0020603fc thread T768 (PUSH-9699)
    #0 0x55f634e64dec in __asan_memcpy (/mnt/hdd01/ci/doris-deploy-branch-3.1-local/be/lib/doris_be+0x39a24dec) (BuildId: 9b04e7f7d3075dac)
    #1 0x55f634eca93f in std::char_traits::copy(char*, char const*, unsigned long) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/char_traits.h:409:33
    #2 0x55f634eca93f in std::__cxx11::basic_string, std::allocator>::_S_copy(char*, char const*, unsigned long) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:351:4
    #3 0x55f634eca93f in std::__cxx11::basic_string, std::allocator>::_S_copy_chars(char*, char const*, char const*) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:398:9
    #4 0x55f634eca93f in void std::__cxx11::basic_string, std::allocator>::_M_construct(char const*, char const*, std::forward_iterator_tag) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.tcc:225:6
    #5 0x55f654a4f74d in void std::__cxx11::basic_string, std::allocator>::_M_construct_aux(char const*, char const*, std::__false_type) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:247:11
    #6 0x55f654a4f74d in void std::__cxx11::basic_string, std::allocator>::_M_construct(char const*, char const*) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:266:4
    #7 0x55f654a4f74d in std::__cxx11::basic_string, std::allocator>::basic_string(char const*, unsigned long, std::allocator const&) /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:513:9
    #8 0x55f654a4f74d in doris::vectorized::parse_thrift_footer(std::shared_ptr, doris::vectorized::FileMetaData**, unsigned long*, doris::io::IOContext*) /home/zcp/repo_center/doris_branch-3.1/doris/be/src/vec/exec/format/parquet/parquet_thrift_util.h:55:17
```

Author: hubgeter

be/src/vec/exec/format/parquet/parquet_thrift_util.h

@@ -48,8 +48,11 @@ static Status parse_thrift_footer(io::FileReaderSPtr file,
 
     // validate magic
     uint8_t* magic_ptr = footer.data() + bytes_read - 4;
-    if (bytes_read < PARQUET_FOOTER_SIZE ||
-        memcmp(magic_ptr, PARQUET_VERSION_NUMBER, sizeof(PARQUET_VERSION_NUMBER)) != 0) {
+    if (bytes_read < PARQUET_FOOTER_SIZE) {
+        return Status::Corruption(
+                "Read parquet file footer fail, bytes read: {}, file size: {}, path: {}",
+                bytes_read, file_size, file->path().native());
+    } else if (memcmp(magic_ptr, PARQUET_VERSION_NUMBER, sizeof(PARQUET_VERSION_NUMBER)) != 0) {
         return Status::Corruption(
                 "Invalid magic number in parquet file, bytes read: {}, file size: {}, path: {}, "
                 "read magic: {}",

regression-test/data/load_p0/ingestion_load/test_ingestion_load_alter_partition.out

@@ -0,0 +1,7 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+2024-09-01	5
+2024-09-02	1
+2024-09-03	1
+2024-09-04	3
+

regression-test/suites/load_p0/ingestion_load/test_ingestion_load.groovy

@@ -21,7 +21,7 @@ import java.nio.file.StandardCopyOption
 
 suite('test_ingestion_load', 'p0,external') {
 
-    def testIngestLoadJob = { testTable, loadLabel, String dataFile ->
+    def testIngestLoadJob = { testTable, loadLabel, String dataFile , filesize ->
 
         sql "TRUNCATE TABLE ${testTable}"
 
@@ -85,7 +85,7 @@ suite('test_ingestion_load', 'p0,external') {
                     "msg": "",
                     "appId": "",
                     "dppResult": "${dppResult}",
-                    "filePathToSize": "{\\"${etlResultFilePath}\\": 81758}",
+                    "filePathToSize": "{\\"${etlResultFilePath}\\": ${filesize}}",
                     "hadoopProperties": "{\\"fs.defaultFS\\":\\"${getHdfsFs()}\\",\\"hadoop.username\\":\\"${getHdfsUser()}\\",\\"hadoop.password\\":\\"${getHdfsPasswd()}\\"}"
                 }
             }"""
@@ -156,7 +156,7 @@ suite('test_ingestion_load', 'p0,external') {
 
         def label = "test_ingestion_load"
 
-        testIngestLoadJob.call(tableName, label, context.config.dataPath + '/load_p0/ingestion_load/data.parquet')
+        testIngestLoadJob.call(tableName, label, context.config.dataPath + '/load_p0/ingestion_load/data.parquet',5745)
 
         tableName = 'tbl_test_spark_load_unique_mor'
 
@@ -189,7 +189,7 @@ suite('test_ingestion_load', 'p0,external') {
 
         label = "test_ingestion_load_unique_mor"
 
-        testIngestLoadJob.call(tableName, label, context.config.dataPath + '/load_p0/ingestion_load/data.parquet')
+        testIngestLoadJob.call(tableName, label, context.config.dataPath + '/load_p0/ingestion_load/data.parquet',5745)
 
         tableName = 'tbl_test_spark_load_agg'
 
@@ -215,7 +215,7 @@ suite('test_ingestion_load', 'p0,external') {
 
         label = "test_ingestion_load_agg"
 
-        testIngestLoadJob.call(tableName, label, context.config.dataPath + '/load_p0/ingestion_load/data1.parquet')
+        testIngestLoadJob.call(tableName, label, context.config.dataPath + '/load_p0/ingestion_load/data1.parquet',4057)
 
     }
 

regression-test/suites/load_p0/ingestion_load/test_ingestion_load_alter_column.groovy

@@ -85,7 +85,7 @@ suite('test_ingestion_load_alter_column', 'p0,external') {
                     "msg": "",
                     "appId": "",
                     "dppResult": "${dppResult}",
-                    "filePathToSize": "{\\"${etlResultFilePath}\\": 81758}",
+                    "filePathToSize": "{\\"${etlResultFilePath}\\": 5745}",
                     "hadoopProperties": "{\\"fs.defaultFS\\":\\"${getHdfsFs()}\\",\\"hadoop.username\\":\\"${getHdfsUser()}\\",\\"hadoop.password\\":\\"${getHdfsPasswd()}\\"}"
                 }
             }"""
@@ -112,7 +112,7 @@ suite('test_ingestion_load_alter_column', 'p0,external') {
         while (max_try_milli_secs) {
             def result = sql "show load where label = '${loadLabel}'"
             if (result[0][2] == "CANCELLED") {
-                msg = result[0][7]
+                def  msg = result[0][7]
                 logger.info("err msg: " + msg)
                 assertTrue((result[0][7] =~ /schema of index \[\d+\] has changed/).find())
                 break
@@ -134,6 +134,8 @@ suite('test_ingestion_load_alter_column', 'p0,external') {
 
         try {
 
+            sql "DROP TABLE if exists ${tableName1}"
+            sql "DROP TABLE if exists ${tableName2}"
             sql """
                 CREATE TABLE IF NOT EXISTS ${tableName1} (
                     c_int int(11) NULL,
@@ -199,10 +201,8 @@ suite('test_ingestion_load_alter_column', 'p0,external') {
             })
 
         } finally {
-            //sql "DROP TABLE ${tableName1}"
-            //sql "DROP TABLE ${tableName2}"
-        }
 
+        }
     }
 
-}
+}

regression-test/suites/load_p0/ingestion_load/test_ingestion_load_alter_partition.groovy

@@ -123,8 +123,8 @@ suite('test_ingestion_load_alter_partition', 'p0,external') {
                 qt_select "select c1, count(*) from ${testTable} group by c1 order by c1"
                 break
             } else if (result[0][2] == "CANCELLED") {
-                msg = result[0][7]
-                logger.info("err msg: " + msg)
+                def msg2 = result[0][7]
+                logger.info("err msg: " + msg2)
                 assertTrue((result[0][7] =~ /partition does not exist/).find())
                 break
             } else {
@@ -146,6 +146,10 @@ suite('test_ingestion_load_alter_partition', 'p0,external') {
 
         try {
 
+            sql "DROP TABLE if exists ${tableName1}"
+            sql "DROP TABLE if exists ${tableName2}"
+            sql "DROP TABLE if exists ${tableName3}"
+
             sql """
                 CREATE TABLE IF NOT EXISTS ${tableName1} (
                     c0 int not null,
@@ -214,9 +218,6 @@ suite('test_ingestion_load_alter_partition', 'p0,external') {
             })
 
         } finally {
-//            sql "DROP TABLE ${tableName1}"
-//            sql "DROP TABLE ${tableName2}"
-//            sql "DROP TABLE ${tableName3}"
         }
 
     }

regression-test/suites/load_p0/ingestion_load/test_ingestion_load_drop_table.groovy

@@ -85,7 +85,7 @@ suite('test_ingestion_load_drop_table', 'p0,external') {
                     "msg": "",
                     "appId": "",
                     "dppResult": "${dppResult}",
-                    "filePathToSize": "{\\"${etlResultFilePath}\\": 81758}",
+                    "filePathToSize": "{\\"${etlResultFilePath}\\": 5745}",
                     "hadoopProperties": "{\\"fs.defaultFS\\":\\"${getHdfsFs()}\\",\\"hadoop.username\\":\\"${getHdfsUser()}\\",\\"hadoop.password\\":\\"${getHdfsPasswd()}\\"}"
                 }
             }"""
@@ -188,7 +188,6 @@ suite('test_ingestion_load_drop_table', 'p0,external') {
             })
 
         } finally {
-            sql "DROP TABLE ${tableName}"
         }
 
     }

regression-test/suites/load_p0/ingestion_load/test_ingestion_load_multi_table.groovy

@@ -103,7 +103,7 @@ suite('test_ingestion_load_multi_table', 'p0,external') {
                     "msg": "",
                     "appId": "",
                     "dppResult": "${dppResult}",
-                    "filePathToSize": "{\\"${etlResultFilePath1}\\": 81758, \\"${etlResultFilePath2}\\": 81758}",
+                    "filePathToSize": "{\\"${etlResultFilePath1}\\": 5745, \\"${etlResultFilePath2}\\": 5745}",
                     "hadoopProperties": "{\\"fs.defaultFS\\":\\"${getHdfsFs()}\\",\\"hadoop.username\\":\\"${getHdfsUser()}\\",\\"hadoop.password\\":\\"${getHdfsPasswd()}\\"}"
                 }
             }"""

regression-test/suites/load_p0/ingestion_load/test_ingestion_load_with_inverted_index.groovy

@@ -85,7 +85,7 @@ suite('test_ingestion_load_with_inverted_index', 'p0,external') {
                     "msg": "",
                     "appId": "",
                     "dppResult": "${dppResult}",
-                    "filePathToSize": "{\\"${etlResultFilePath}\\": 81758}",
+                    "filePathToSize": "{\\"${etlResultFilePath}\\": 5745}",
                     "hadoopProperties": "{\\"fs.defaultFS\\":\\"${getHdfsFs()}\\",\\"hadoop.username\\":\\"${getHdfsUser()}\\",\\"hadoop.password\\":\\"${getHdfsPasswd()}\\"}"
                 }
             }"""

regression-test/suites/load_p0/ingestion_load/test_ingestion_load_with_partition.groovy

@@ -71,7 +71,7 @@ suite('test_ingestion_load_with_partition', 'p0,external') {
             }
         }
 
-        etlResultFilePaths = []
+        def etlResultFilePaths = []
         for(int i=0; i < dataFiles.size(); i++) {
             Files.copy(Paths.get(dataFiles[i]),
                 Paths.get(context.config.dataPath + "/load_p0/ingestion_load/${resultFileNames[i]}"), StandardCopyOption.REPLACE_EXISTING)
@@ -115,7 +115,7 @@ suite('test_ingestion_load_with_partition', 'p0,external') {
 
         def max_try_milli_secs = 120000
         while (max_try_milli_secs) {
-            result = sql "show load where label = '${loadLabel}'"
+            def result = sql "show load where label = '${loadLabel}'"
             if (result[0][2] == "FINISHED") {
                 sql "sync"
                 qt_select "select c1, count(*) from ${testTable} group by c1 order by c1"
@@ -132,9 +132,8 @@ suite('test_ingestion_load_with_partition', 'p0,external') {
     }
 
     if (enableHdfs()) {
-
-        tableName = 'tbl_test_spark_load_partition'
-
+        def tableName = 'tbl_test_spark_load_with_partition'
+        sql "DROP TABLE if exists ${tableName}"
         sql """
             CREATE TABLE IF NOT EXISTS ${tableName} (
                 c0 int not null,
@@ -151,10 +150,10 @@ suite('test_ingestion_load_with_partition', 'p0,external') {
             )
             """
 
-        def label = "test_ingestion_load_partition"
+        def label = "test_ingestion_load_with_partition__"
 
         testIngestLoadJob.call(tableName, label, [context.config.dataPath + '/load_p0/ingestion_load/data2-0.parquet', context.config.dataPath + '/load_p0/ingestion_load/data2-1.parquet',context.config.dataPath + '/load_p0/ingestion_load/data2-2.parquet',context.config.dataPath + '/load_p0/ingestion_load/data2-3.parquet'])
 
     }
 
-}
+}