branch-2.1: [fix](tvf) support compressed json file for tvf and refactor code (#51983) (#53063)

bp: #51983

Author: suxiaogang223

be/src/vec/exec/format/json/new_json_reader.cpp

@@ -248,7 +248,9 @@ Status NewJsonReader::get_columns(std::unordered_map<std::string, TypeDescriptor
 Status NewJsonReader::get_parsed_schema(std::vector<std::string>* col_names,
                                         std::vector<TypeDescriptor>* col_types) {
     RETURN_IF_ERROR(_get_range_params());
-
+    // create decompressor.
+    // _decompressor may be nullptr if this is not a compressed file
+    RETURN_IF_ERROR(Decompressor::create_decompressor(_file_compress_type, &_decompressor));
     RETURN_IF_ERROR(_open_file_reader(true));
     if (_read_json_by_line) {
         RETURN_IF_ERROR(_open_line_reader());

docker/thirdparties/docker-compose/hive/scripts/preinstalled_data/json_format_test/simple_object_json.json.gz

@@ -286,6 +286,20 @@
 8	chengdu	2345678
 9	xian	2345679
 
+-- !json_compressed --
+1	beijing	2345671
+10	hefei	23456710
+11	\N	23456711
+12	hefei	\N
+2	shanghai	2345672
+3	guangzhou	2345673
+4	shenzhen	2345674
+5	hangzhou	2345675
+6	nanjing	2345676
+7	wuhan	2345677
+8	chengdu	2345678
+9	xian	2345679
+
 -- !json_limit1 --
 1	beijing	2345671
 10	hefei	23456710

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

@@ -143,6 +143,16 @@ suite("test_hdfs_tvf","external,hive,tvf,external_docker") {
                         "strip_outer_array" = "false",
                         "read_json_by_line" = "true") order by id; """
 
+            uri = "${defaultFS}" + "/user/doris/preinstalled_data/json_format_test/simple_object_json.json.gz"
+            format = "json"
+            qt_json_compressed """ select * from HDFS(
+                        "uri" = "${uri}",
+                        "hadoop.username" = "${hdfsUserName}",
+                        "format" = "${format}",
+                        "compress_type" = "GZ",
+                        "strip_outer_array" = "false",
+                        "read_json_by_line" = "true") order by id; """
+
 
            uri = "${defaultFS}" + "/user/doris/preinstalled_data/json_format_test/simple_object_json.json"
             format = "json"