Merge branch 'master' into master-fix1

SWJTU-ZhangLei · web-flow · commit 2046a2627074 · 2024-11-01T16:42:32.000+08:00
diff --git a/.asf.yaml b/.asf.yaml
@@ -152,13 +152,13 @@ github:
     - LemonLiTree
     - Yukang-Lian
     - TangSiyang2001
-    - Lchangliang
     - freemandealer
     - shuke987
     - wm1581066
     - KassieZ
     - yujun777
     - doris-robot
+    - LiBinfeng-01
 
 notifications:
     pullrequests_status:  commits@doris.apache.org
diff --git a/be/src/cloud/cloud_tablet.cpp b/be/src/cloud/cloud_tablet.cpp
@@ -108,6 +108,36 @@ Status CloudTablet::capture_rs_readers(const Version& spec_version,
     return capture_rs_readers_unlocked(version_path, rs_splits);
 }
 
+Status CloudTablet::merge_rowsets_schema() {
+    // Find the rowset with the max version
+    auto max_version_rowset =
+            std::max_element(
+                    _rs_version_map.begin(), _rs_version_map.end(),
+                    [](const auto& a, const auto& b) {
+                        return !a.second->tablet_schema()
+                                       ? true
+                                       : (!b.second->tablet_schema()
+                                                  ? false
+                                                  : a.second->tablet_schema()->schema_version() <
+                                                            b.second->tablet_schema()
+                                                                    ->schema_version());
+                    })
+                    ->second;
+    TabletSchemaSPtr max_version_schema = max_version_rowset->tablet_schema();
+    // If the schema has variant columns, perform a merge to create a wide tablet schema
+    if (max_version_schema->num_variant_columns() > 0) {
+        std::vector<TabletSchemaSPtr> schemas;
+        std::transform(_rs_version_map.begin(), _rs_version_map.end(), std::back_inserter(schemas),
+                       [](const auto& rs_meta) { return rs_meta.second->tablet_schema(); });
+        // Merge the collected schemas to obtain the least common schema
+        RETURN_IF_ERROR(vectorized::schema_util::get_least_common_schema(schemas, nullptr,
+                                                                         max_version_schema));
+        VLOG_DEBUG << "dump schema: " << max_version_schema->dump_full_schema();
+        _merged_tablet_schema = max_version_schema;
+    }
+    return Status::OK();
+}
+
 // There are only two tablet_states RUNNING and NOT_READY in cloud mode
 // This function will erase the tablet from `CloudTabletMgr` when it can't find this tablet in MS.
 Status CloudTablet::sync_rowsets(int64_t query_version, bool warmup_delta_data) {
@@ -133,6 +163,10 @@ Status CloudTablet::sync_rowsets(int64_t query_version, bool warmup_delta_data)
     if (st.is<ErrorCode::NOT_FOUND>()) {
         clear_cache();
     }
+
+    // Merge all rowset schemas within a CloudTablet
+    RETURN_IF_ERROR(merge_rowsets_schema());
+
     return st;
 }
 
@@ -188,16 +222,7 @@ Status CloudTablet::sync_if_not_running() {
 }
 
 TabletSchemaSPtr CloudTablet::merged_tablet_schema() const {
-    std::shared_lock rdlock(_meta_lock);
-    TabletSchemaSPtr target_schema;
-    std::vector<TabletSchemaSPtr> schemas;
-    for (const auto& [_, rowset] : _rs_version_map) {
-        schemas.push_back(rowset->tablet_schema());
-    }
-    // get the max version schema and merge all schema
-    static_cast<void>(
-            vectorized::schema_util::get_least_common_schema(schemas, nullptr, target_schema));
-    return target_schema;
+    return _merged_tablet_schema;
 }
 
 void CloudTablet::add_rowsets(std::vector<RowsetSharedPtr> to_add, bool version_overlap,
diff --git a/be/src/cloud/cloud_tablet.h b/be/src/cloud/cloud_tablet.h
@@ -208,6 +208,9 @@ class CloudTablet final : public BaseTablet {
 
     Status sync_if_not_running();
 
+    // Merge all rowset schemas within a CloudTablet
+    Status merge_rowsets_schema();
+
     CloudStorageEngine& _engine;
 
     // this mutex MUST ONLY be used when sync meta
@@ -246,6 +249,9 @@ class CloudTablet final : public BaseTablet {
     std::mutex _base_compaction_lock;
     std::mutex _cumulative_compaction_lock;
     mutable std::mutex _rowset_update_lock;
+
+    // Schema will be merged from all rowsets when sync_rowsets
+    TabletSchemaSPtr _merged_tablet_schema;
 };
 
 using CloudTabletSPtr = std::shared_ptr<CloudTablet>;
diff --git a/be/src/http/action/adjust_log_level.cpp b/be/src/http/action/adjust_log_level.cpp
@@ -17,16 +17,17 @@
 
 #include <http/action/adjust_log_level.h>
 
+#include <tuple>
+
 #include "common/logging.h"
-#include "common/status.h"
 #include "http/http_channel.h"
 #include "http/http_request.h"
 
 namespace doris {
 
 // **Note**: If the module_name does not exist in the vlog modules, vlog
 // would create corresponding module for it.
-int handle_request(HttpRequest* req) {
+std::tuple<std::string, int, int> handle_request(HttpRequest* req) {
     auto parse_param = [&req](std::string param) {
         const auto& value = req->param(param);
         if (value.empty()) {
@@ -38,13 +39,16 @@ int handle_request(HttpRequest* req) {
     const auto& module = parse_param("module");
     const auto& level = parse_param("level");
     int new_level = std::stoi(level);
-    return google::SetVLOGLevel(module.c_str(), new_level);
+    return std::make_tuple(module, google::SetVLOGLevel(module.c_str(), new_level), new_level);
 }
 
 void AdjustLogLevelAction::handle(HttpRequest* req) {
     try {
-        auto old_level = handle_request(req);
-        auto msg = fmt::format("adjust log level success, origin level is {}", old_level);
+        auto handle_result = handle_request(req);
+        auto msg =
+                fmt::format("adjust vlog of {} from {} to {} succeed", std::get<0>(handle_result),
+                            std::get<1>(handle_result), std::get<2>(handle_result));
+        LOG(INFO) << msg;
         HttpChannel::send_reply(req, msg);
     } catch (const std::exception& e) {
         HttpChannel::send_reply(req, HttpStatus::INTERNAL_SERVER_ERROR, e.what());
diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp
@@ -1159,7 +1159,10 @@ void PInternalService::fetch_remote_tablet_schema(google::protobuf::RpcControlle
                         LOG(WARNING) << "tablet does not exist, tablet id is " << tablet_id;
                         continue;
                     }
-                    tablet_schemas.push_back(res.value()->merged_tablet_schema());
+                    auto schema = res.value()->merged_tablet_schema();
+                    if (schema != nullptr) {
+                        tablet_schemas.push_back(schema);
+                    }
                 }
                 if (!tablet_schemas.empty()) {
                     // merge all
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/TablePartitionValues.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/TablePartitionValues.java
@@ -34,11 +34,7 @@
 import com.google.common.collect.RangeMap;
 import lombok.Data;
 
-import java.io.UnsupportedEncodingException;
-import java.net.URLDecoder;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -80,11 +76,6 @@ public TablePartitionValues(List<String> partitionNames, List<List<String>> part
         addPartitions(partitionNames, partitionValues, types);
     }
 
-    public TablePartitionValues(List<String> partitionNames, List<Type> types) {
-        this();
-        addPartitions(partitionNames, types);
-    }
-
     public void addPartitions(List<String> partitionNames, List<List<String>> partitionValues, List<Type> types) {
         Preconditions.checkState(partitionNames.size() == partitionValues.size());
         List<String> addPartitionNames = new ArrayList<>();
@@ -105,10 +96,6 @@ public void addPartitions(List<String> partitionNames, List<List<String>> partit
         addPartitionItems(addPartitionNames, addPartitionItems, types);
     }
 
-    public void addPartitions(List<String> partitionNames, List<Type> types) {
-        addPartitions(partitionNames,
-                partitionNames.stream().map(this::getHivePartitionValues).collect(Collectors.toList()), types);
-    }
 
     private void addPartitionItems(List<String> partitionNames, List<PartitionItem> partitionItems, List<Type> types) {
         Preconditions.checkState(partitionNames.size() == partitionItems.size());
@@ -196,23 +183,6 @@ private ListPartitionItem toListPartitionItem(List<String> partitionValues, List
         }
     }
 
-    private List<String> getHivePartitionValues(String partitionName) {
-        // Partition name will be in format: nation=cn/city=beijing
-        // parse it to get values "cn" and "beijing"
-        return Arrays.stream(partitionName.split("/")).map(part -> {
-            String[] kv = part.split("=");
-            Preconditions.checkState(kv.length == 2, partitionName);
-            String partitionValue;
-            try {
-                // hive partition value maybe contains special characters like '=' and '/'
-                partitionValue = URLDecoder.decode(kv[1], StandardCharsets.UTF_8.name());
-            } catch (UnsupportedEncodingException e) {
-                // It should not be here
-                throw new RuntimeException(e);
-            }
-            return partitionValue;
-        }).collect(Collectors.toList());
-    }
 
     @Data
     public static class TablePartitionKey {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveMetaStoreCache.java
@@ -244,7 +244,7 @@ public Long getValue() {
     }
 
     private HivePartitionValues loadPartitionValues(PartitionValueCacheKey key) {
-        // partition name format: nation=cn/city=beijing
+        // partition name format: nation=cn/city=beijing,`listPartitionNames` returned string is the encoded string.
         List<String> partitionNames = catalog.getClient().listPartitionNames(key.dbName, key.tblName);
         if (LOG.isDebugEnabled()) {
             LOG.debug("load #{} partitions for {} in catalog {}", partitionNames.size(), key, catalog.getName());
@@ -281,11 +281,10 @@ private HivePartitionValues loadPartitionValues(PartitionValueCacheKey key) {
     public ListPartitionItem toListPartitionItem(String partitionName, List<Type> types) {
         // Partition name will be in format: nation=cn/city=beijing
         // parse it to get values "cn" and "beijing"
-        String[] parts = partitionName.split("/");
-        Preconditions.checkState(parts.length == types.size(), partitionName + " vs. " + types);
+        List<String> partitionValues = HiveUtil.toPartitionValues(partitionName);
+        Preconditions.checkState(partitionValues.size() == types.size(), partitionName + " vs. " + types);
         List<PartitionValue> values = Lists.newArrayListWithExpectedSize(types.size());
-        for (String part : parts) {
-            String partitionValue = HiveUtil.getHivePartitionValue(part);
+        for (String partitionValue : partitionValues) {
             values.add(new PartitionValue(partitionValue, HIVE_DEFAULT_PARTITION.equals(partitionValue)));
         }
         try {
@@ -325,9 +324,9 @@ private Map<PartitionCacheKey, HivePartition> loadPartitions(Iterable<? extends
             StringBuilder sb = new StringBuilder();
             Preconditions.checkState(key.getValues().size() == partitionColumns.size());
             for (int i = 0; i < partitionColumns.size(); i++) {
-                sb.append(partitionColumns.get(i).getName());
+                // Partition name and value  may contain special character, like / and so on. Need to encode.
+                sb.append(FileUtils.escapePathName(partitionColumns.get(i).getName()));
                 sb.append("=");
-                // Partition value may contain special character, like / and so on. Need to encode.
                 sb.append(FileUtils.escapePathName(key.getValues().get(i)));
                 sb.append("/");
             }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveUtil.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HiveUtil.java
@@ -49,9 +49,6 @@
 import org.apache.hadoop.mapred.TextInputFormat;
 import org.apache.hadoop.util.ReflectionUtils;
 
-import java.io.UnsupportedEncodingException;
-import java.net.URLDecoder;
-import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -123,16 +120,22 @@ public static boolean isSplittable(RemoteFileSystem remoteFileSystem, String inp
         return HMSExternalTable.SUPPORTED_HIVE_FILE_FORMATS.contains(inputFormat);
     }
 
-    public static String getHivePartitionValue(String part) {
-        String[] kv = part.split("=");
-        Preconditions.checkState(kv.length == 2, String.format("Malformed partition name %s", part));
-        try {
-            // hive partition value maybe contains special characters like '=' and '/'
-            return URLDecoder.decode(kv[1], StandardCharsets.UTF_8.name());
-        } catch (UnsupportedEncodingException e) {
-            // It should not be here
-            throw new RuntimeException(e);
+    // "c1=a/c2=b/c3=c" ---> List(["c1","a"], ["c2","b"], ["c3","c"])
+    // Similar to the `toPartitionValues` method, except that it adds the partition column name.
+    public static List<String[]> toPartitionColNameAndValues(String partitionName) {
+
+        String[] parts = partitionName.split("/");
+        List<String[]> result = new ArrayList<>(parts.length);
+        for (String part : parts) {
+            String[] kv = part.split("=");
+            Preconditions.checkState(kv.length == 2, String.format("Malformed partition name %s", part));
+
+            result.add(new String[] {
+                    FileUtils.unescapePathName(kv[0]),
+                    FileUtils.unescapePathName(kv[1])
+            });
         }
+        return result;
     }
 
     // "c1=a/c2=b/c3=c" ---> List("a","b","c")
@@ -151,6 +154,8 @@ public static List<String> toPartitionValues(String partitionName) {
             if (start > partitionName.length()) {
                 break;
             }
+            //Ref: common/src/java/org/apache/hadoop/hive/common/FileUtils.java
+            //makePartName(List<String> partCols, List<String> vals,String defaultStr)
             resultBuilder.add(FileUtils.unescapePathName(partitionName.substring(start, end)));
             start = end + 1;
         }
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
@@ -1933,6 +1933,11 @@ private void handleShowHMSTablePartitions(ShowPartitionsStmt showStmt) throws An
         Map<String, Expr> filterMap = showStmt.getFilterMap();
         List<OrderByPair> orderByPairs = showStmt.getOrderByPairs();
 
+        // catalog.getClient().listPartitionNames() returned string is the encoded string.
+        // example: insert into tmp partition(pt="1=3/3") values( xxx );
+        //          show partitions from tmp: pt=1%3D3%2F3
+        // Need to consider whether to call `HiveUtil.toPartitionColNameAndValues` method
+
         if (limit != null && limit.hasLimit() && limit.getOffset() == 0
                 && (orderByPairs == null || !orderByPairs.get(0).isDesc())) {
             // hmsClient returns unordered partition list, hence if offset > 0 cannot pass limit
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -1912,6 +1912,7 @@ public void executeAndSendResult(boolean isOutfileQuery, boolean isSendFields,
                                 : new ShortCircuitQueryContext(planner, (Queriable) parsedStmt);
             coordBase = new PointQueryExecutor(shortCircuitQueryContext,
                         context.getSessionVariable().getMaxMsgSizeOfResultReceiver());
+            context.getState().setIsQuery(true);
         } else if (planner instanceof NereidsPlanner && ((NereidsPlanner) planner).getDistributedPlans() != null) {
             coord = new NereidsCoordinator(context, analyzer,
                     planner, context.getStatsErrorEstimator(),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/HMSAnalysisTask.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/HMSAnalysisTask.java
@@ -104,10 +104,11 @@ private void getPartitionColumnStats() {
         for (String names : partitionNames) {
             // names is like "date=20230101" for one level partition
             // and like "date=20230101/hour=12" for two level partition
-            String[] parts = names.split("/");
-            for (String part : parts) {
-                if (part.startsWith(col.getName())) {
-                    String value = HiveUtil.getHivePartitionValue(part);
+            List<String[]> parts = HiveUtil.toPartitionColNameAndValues(names);
+            for (String[] part : parts) {
+                String colName = part[0];
+                String value = part[1];
+                if (colName != null && colName.equals(col.getName())) {
                     // HIVE_DEFAULT_PARTITION hive partition value when the partition name is not specified.
                     if (value == null || value.isEmpty() || value.equals(HiveMetaStoreCache.HIVE_DEFAULT_PARTITION)) {
                         numNulls += 1;
diff --git a/regression-test/data/external_table_p0/hive/test_hive_special_char_partition.out b/regression-test/data/external_table_p0/hive/test_hive_special_char_partition.out
diff --git a/regression-test/suites/external_table_p0/hive/test_hive_special_char_partition.groovy b/regression-test/suites/external_table_p0/hive/test_hive_special_char_partition.groovy

Original file line number	Diff line number	Diff line change
`@@ -1159,7 +1159,10 @@ void PInternalService::fetch_remote_tablet_schema(google::protobuf::RpcControlle`
`1159`	`1159`	`LOG(WARNING) << "tablet does not exist, tablet id is " << tablet_id;`
`1160`	`1160`	`continue;`
`1161`	`1161`	`}`
`1162`		`- tablet_schemas.push_back(res.value()->merged_tablet_schema());`
	`1162`	`+ auto schema = res.value()->merged_tablet_schema();`
	`1163`	`+ if (schema != nullptr) {`
	`1164`	`+ tablet_schemas.push_back(schema);`
	`1165`	`+ }`
`1163`	`1166`	`}`
`1164`	`1167`	`if (!tablet_schemas.empty()) {`
`1165`	`1168`	`// merge all`