[Feature] Add flat_json.column_paths to control which JSON paths are …

be/src/storage/flat_json_config.h

@@ -17,12 +17,21 @@
 #include <gen_cpp/olap_file.pb.h>
 
 #include <sstream>
+#include <string>
+#include <unordered_map>
+#include <unordered_set>
+#include <vector>
 
 #include "gen_cpp/AgentService_types.h"
 
 namespace starrocks {
 class FlatJsonConfig {
 public:
+    // Default max force-path columns (per JSON column) when not specified by the caller.
+    static constexpr int DEFAULT_COLUMN_PATHS_MAX = 200;
+
+    using ColumnPathsMap = std::unordered_map<std::string, std::unordered_set<std::string>>;
+
     // Constructor
     FlatJsonConfig();
 
@@ -31,7 +40,8 @@ class FlatJsonConfig {
             : _flat_json_enable(enable),
               _flat_json_null_factor(nullFactor),
               _flat_json_sparsity_factor(sparsityFactor),
-              _flat_json_max_column_max(maxColumnMax) {}
+              _flat_json_max_column_max(maxColumnMax),
+              _flat_json_column_paths_max(DEFAULT_COLUMN_PATHS_MAX) {}
 
     // Getters and Setters
     bool is_flat_json_enabled() const { return _flat_json_enable; }
@@ -46,27 +56,96 @@ class FlatJsonConfig {
     int get_flat_json_max_column_max() const { return _flat_json_max_column_max; }
     void set_flat_json_max_column_max(int max) { _flat_json_max_column_max = max; }
 
+    // Per-JSON-column force-flatten paths (dot-separated, no leading "$.").
+    const ColumnPathsMap& get_column_paths_map() const { return _flat_json_column_paths; }
+
+    // Returns the path set for the given JSON column, or nullptr if the column has no forced paths.
+    const std::unordered_set<std::string>* get_column_paths_for(const std::string& column_name) const {
+        auto it = _flat_json_column_paths.find(column_name);
+        return it == _flat_json_column_paths.end() ? nullptr : &it->second;
+    }
+
+    void set_column_paths_map(ColumnPathsMap paths) { _flat_json_column_paths = std::move(paths); }
+
+    void set_column_paths(const std::string& column_name, const std::vector<std::string>& paths) {
+        std::unordered_set<std::string> s;
+        for (const auto& p : paths) {
+            s.insert(p);
+        }
+        _flat_json_column_paths[column_name] = std::move(s);
+    }
+
+    int get_column_paths_max() const { return _flat_json_column_paths_max; }
+    void set_column_paths_max(int max) { _flat_json_column_paths_max = max; }
+
     void to_pb(FlatJsonConfigPB* binlog_config_pb) {
         binlog_config_pb->set_flat_json_enable(_flat_json_enable);
         binlog_config_pb->set_flat_json_null_factor(_flat_json_null_factor);
         binlog_config_pb->set_flat_json_sparsity_factor(_flat_json_sparsity_factor);
         binlog_config_pb->set_flat_json_max_column_max(_flat_json_max_column_max);
+        binlog_config_pb->clear_flat_json_column_paths();
+        for (const auto& [col, paths] : _flat_json_column_paths) {
+            auto* entry = binlog_config_pb->add_flat_json_column_paths();
+            entry->set_column_name(col);
+            for (const auto& p : paths) {
+                entry->add_paths(p);
+            }
+        }
+        binlog_config_pb->set_flat_json_column_paths_max(_flat_json_column_paths_max);
     }
 
     // Update function using another FlatJsonConfig
     void update(const FlatJsonConfig& config) {
-        update(config.is_flat_json_enabled(), config.get_flat_json_null_factor(),
-               config.get_flat_json_sparsity_factor(), config.get_flat_json_max_column_max());
+        _flat_json_enable = config.is_flat_json_enabled();
+        _flat_json_null_factor = config.get_flat_json_null_factor();
+        _flat_json_sparsity_factor = config.get_flat_json_sparsity_factor();
+        _flat_json_max_column_max = config.get_flat_json_max_column_max();
+        _flat_json_column_paths = config.get_column_paths_map();
+        _flat_json_column_paths_max = config.get_column_paths_max();
     }
 
     void update(const TFlatJsonConfig& config) {
-        update(config.flat_json_enable, config.flat_json_null_factor, config.flat_json_sparsity_factor,
-               config.flat_json_column_max);
+        _flat_json_enable = config.flat_json_enable;
+        _flat_json_null_factor = config.flat_json_null_factor;
+        _flat_json_sparsity_factor = config.flat_json_sparsity_factor;
+        _flat_json_max_column_max = config.flat_json_column_max;
+        _flat_json_column_paths.clear();
+        if (config.__isset.flat_json_column_paths) {
+            for (const auto& [col, paths] : config.flat_json_column_paths) {
+                std::unordered_set<std::string> s(paths.begin(), paths.end());
+                _flat_json_column_paths.emplace(col, std::move(s));
+            }
+        }
+        if (config.__isset.flat_json_column_paths_max && config.flat_json_column_paths_max > 0) {
+            _flat_json_column_paths_max = static_cast<int>(config.flat_json_column_paths_max);
+        }
     }
 
     void update(const FlatJsonConfigPB& flat_json_config_pb) {
-        update(flat_json_config_pb.flat_json_enable(), flat_json_config_pb.flat_json_null_factor(),
-               flat_json_config_pb.flat_json_sparsity_factor(), flat_json_config_pb.flat_json_max_column_max());
+        _flat_json_enable = flat_json_config_pb.flat_json_enable();
+        _flat_json_null_factor = flat_json_config_pb.flat_json_null_factor();
+        _flat_json_sparsity_factor = flat_json_config_pb.flat_json_sparsity_factor();
+        _flat_json_max_column_max = flat_json_config_pb.flat_json_max_column_max();
+        _flat_json_column_paths.clear();
+        for (const auto& entry : flat_json_config_pb.flat_json_column_paths()) {
+            std::unordered_set<std::string> s;
+            for (const auto& p : entry.paths()) {
+                s.insert(p);
+            }
+            _flat_json_column_paths.emplace(entry.column_name(), std::move(s));
+        }
+        if (flat_json_config_pb.has_flat_json_column_paths_max() &&
+            flat_json_config_pb.flat_json_column_paths_max() > 0) {
+            _flat_json_column_paths_max = static_cast<int>(flat_json_config_pb.flat_json_column_paths_max());
+        }
+    }
+
+    // Update function using four parameters (kept for backward compatibility with original API)
+    void update(bool enable, double nullFactor, double sparsityFactor, int maxColumnMax) {
+        _flat_json_enable = enable;
+        _flat_json_null_factor = nullFactor;
+        _flat_json_sparsity_factor = sparsityFactor;
+        _flat_json_max_column_max = maxColumnMax;
     }
 
     // Copy Assignment
@@ -76,25 +155,35 @@ class FlatJsonConfig {
             _flat_json_null_factor = other._flat_json_null_factor;
             _flat_json_sparsity_factor = other._flat_json_sparsity_factor;
             _flat_json_max_column_max = other._flat_json_max_column_max;
+            _flat_json_column_paths = other._flat_json_column_paths;
+            _flat_json_column_paths_max = other._flat_json_column_paths_max;
         }
         return *this;
     }
 
-    // Update function using four parameters
-    void update(bool enable, double nullFactor, double sparsityFactor, int maxColumnMax) {
-        _flat_json_enable = enable;
-        _flat_json_null_factor = nullFactor;
-        _flat_json_sparsity_factor = sparsityFactor;
-        _flat_json_max_column_max = maxColumnMax;
-    }
-
     std::string to_string() const {
         std::ostringstream oss;
         oss << "FlatJsonConfig{";
         oss << "flat_json_enable=" << (_flat_json_enable ? "true" : "false") << ", ";
         oss << "flat_json_null_factor=" << _flat_json_null_factor << ", ";
         oss << "flat_json_sparsity_factor=" << _flat_json_sparsity_factor << ", ";
-        oss << "flat_json_max_column_max=" << _flat_json_max_column_max;
+        oss << "flat_json_max_column_max=" << _flat_json_max_column_max << ", ";
+        oss << "flat_json_column_paths={";
+        bool first_col = true;
+        for (const auto& [col, paths] : _flat_json_column_paths) {
+            if (!first_col) oss << ",";
+            oss << col << ":[";
+            bool first_p = true;
+            for (const auto& p : paths) {
+                if (!first_p) oss << ",";
+                oss << p;
+                first_p = false;
+            }
+            oss << "]";
+            first_col = false;
+        }
+        oss << "}, ";
+        oss << "flat_json_column_paths_max=" << _flat_json_column_paths_max;
         oss << "}";
         return oss.str();
     }
@@ -104,5 +193,8 @@ class FlatJsonConfig {
     double _flat_json_null_factor = 0;
     double _flat_json_sparsity_factor = 0;
     int _flat_json_max_column_max = 0;
+    // Per-JSON-column force-flatten paths: column_name -> set of dot-separated paths (no leading "$.").
+    ColumnPathsMap _flat_json_column_paths;
+    int _flat_json_column_paths_max = DEFAULT_COLUMN_PATHS_MAX;
 };
 } // namespace starrocks

be/src/storage/lake/tablet_reader.cpp

@@ -269,7 +269,7 @@ Status TabletReader::init_compaction_column_paths(const TabletReaderParams& read
                 metadata && metadata->has_flat_json_config()) {
                 auto flat_json_config = std::make_shared<FlatJsonConfig>();
                 flat_json_config->update(metadata->flat_json_config());
-                deriver.init_flat_json_config(flat_json_config.get());
+                deriver.init_flat_json_config(flat_json_config.get(), col_name);
             }
 
             deriver.derived(readers);

be/src/storage/rowset/json_column_compactor.cpp

@@ -46,7 +46,7 @@ Status FlatJsonColumnCompactor::_compact_columns(MutableColumns& json_datas) {
         vc.emplace_back(js.get());
     }
     deriver.set_generate_filter(true);
-    deriver.init_flat_json_config(_flat_json_config);
+    deriver.init_flat_json_config(_flat_json_config, _column_name);
 
     deriver.derived(vc);
 

be/src/storage/rowset/json_column_writer.cpp

@@ -87,7 +87,7 @@ Status FlatJsonColumnWriter::append(const Column& column) {
 Status FlatJsonColumnWriter::_flat_column(MutableColumns& json_datas) {
     // all json datas must full json
     JsonPathDeriver deriver;
-    deriver.init_flat_json_config(_flat_json_config);
+    deriver.init_flat_json_config(_flat_json_config, _column_name);
     deriver.set_generate_filter(true);
 
     std::vector<const Column*> vc;

be/src/storage/tablet_reader.cpp

@@ -187,7 +187,7 @@ Status TabletReader::_init_compaction_column_paths(const TabletReaderParams& rea
             // must all be flat json type
             JsonPathDeriver deriver;
             auto flat_json_config = _tablet->flat_json_config();
-            deriver.init_flat_json_config(flat_json_config.get());
+            deriver.init_flat_json_config(flat_json_config.get(), col_name);
             deriver.derived(readers);
             auto paths = deriver.flat_paths();
             auto types = deriver.flat_types();

be/src/util/json_flattener.cpp

@@ -391,11 +391,19 @@ JsonPathDeriver::JsonPathDeriver(const std::vector<std::string>& paths, const st
     }
 }
 
-void JsonPathDeriver::init_flat_json_config(const FlatJsonConfig* flat_json_config) {
+void JsonPathDeriver::init_flat_json_config(const FlatJsonConfig* flat_json_config, const std::string& column_name) {
     if (flat_json_config != nullptr) {
         _max_json_null_factor = flat_json_config->get_flat_json_null_factor();
         _min_json_sparsity_factory = flat_json_config->get_flat_json_sparsity_factor();
         _max_column = flat_json_config->get_flat_json_max_column_max();
+        _column_paths.clear();
+        if (!column_name.empty()) {
+            // Load only the paths targeted at this specific JSON column.
+            if (const auto* paths = flat_json_config->get_column_paths_for(column_name); paths != nullptr) {
+                _column_paths = *paths;
+            }
+        }
+        _column_paths_max = flat_json_config->get_column_paths_max();
     } else {
         _max_json_null_factor = config::json_flat_null_factor;
         _min_json_sparsity_factory = config::json_flat_sparsity_factor;
@@ -420,6 +428,10 @@ void JsonPathDeriver::derived(const std::vector<const Column*>& json_datas) {
     _total_rows = res.value();
 
     _path_root = std::make_shared<JsonFlatPath>();
+    // Pre-mark force paths so _clean_sparsity_path never prunes them.
+    for (const auto& fp : _column_paths) {
+        _mark_force_path(fp, _path_root.get());
+    }
     // init path by flat JSON
     _derived_on_flat_json(json_datas);
 
@@ -565,14 +577,30 @@ void JsonPathDeriver::_derived(const Column* col, size_t mark_row) {
     }
 }
 
+// Walk the tree along `path` (dot-separated), creating nodes as needed, and
+// mark every node on the path with force=true so they survive early pruning.
+void JsonPathDeriver::_mark_force_path(const std::string_view& path, JsonFlatPath* node) {
+    node->force = true;
+    if (path.empty()) {
+        return;
+    }
+    auto [key, next] = JsonFlatPath::split_path(path);
+    auto [iter, inserted] = node->children.try_emplace(key);
+    if (inserted) {
+        iter->second = std::make_unique<JsonFlatPath>();
+    }
+    _mark_force_path(next, iter->second.get());
+}
+
 void JsonPathDeriver::_clean_sparsity_path(const std::string_view& name, JsonFlatPath* node, size_t check_hits_min) {
     for (auto& [key, child] : node->children) {
         _clean_sparsity_path(key, child.get(), check_hits_min);
     }
     auto iter = node->children.begin();
     while (iter != node->children.end()) {
         auto child = iter->second.get();
-        if (child->hits < check_hits_min) {
+        // Never prune a node that belongs to a force path.
+        if (child->hits < check_hits_min && !child->force) {
             if (_generate_filter) {
                 _remain_keys.insert(iter->first);
             }
@@ -685,7 +713,17 @@ uint32_t JsonPathDeriver::_dfs_finalize(JsonFlatPath* node, const std::string& a
 
         bool is_base_type = node->base_type_count >= node->hits - (node->hits * config::json_flat_complex_type_factor);
         bool type_check = config::enable_json_flat_complex_type || is_base_type;
-        if (type_check && node->multi_times <= 0 && node->hits >= _total_rows * _min_json_sparsity_factory) {
+
+        // Force paths bypass the sparsity threshold entirely.
+        // We still skip if multi_times > 0 (ambiguous: same key appears twice in
+        // one object) or hits == 0 (path never seen in any row).
+        if (node->force && node->hits > 0 && node->multi_times <= 0) {
+            hit_leaf->emplace_back(node, absolute_path);
+            node->type = flat_json::JSON_BITS_TO_LOGICAL_TYPE.at(node->json_type);
+            node->remain = false;
+            return 1;
+        } else if (!node->force && type_check && node->multi_times <= 0 &&
+                   node->hits >= _total_rows * _min_json_sparsity_factory) {
             hit_leaf->emplace_back(node, absolute_path);
             node->type = flat_json::JSON_BITS_TO_LOGICAL_TYPE.at(node->json_type);
             node->remain = false;
@@ -726,21 +764,51 @@ void JsonPathDeriver::_finalize() {
     std::vector<std::pair<JsonFlatPath*, std::string>> hit_leaf;
     _dfs_finalize(_path_root.get(), "", &hit_leaf);
 
-    // sort by name, just for stable order
-    std::sort(hit_leaf.begin(), hit_leaf.end(),
-              [&](const auto& a, const auto& b) { return a.first->hits > b.first->hits; });
-    size_t limit = _max_column > 0 ? _max_column : std::numeric_limits<size_t>::max();
-    for (size_t i = limit; i < hit_leaf.size(); i++) {
-        if (!hit_leaf[i].first->remain && hit_leaf[i].first->hits >= _total_rows) {
+    // Separate forced leaves from normal leaves so each has its own quota.
+    std::vector<std::pair<JsonFlatPath*, std::string>> forced_leaves;
+    std::vector<std::pair<JsonFlatPath*, std::string>> normal_leaves;
+    for (auto& item : hit_leaf) {
+        if (item.first->force) {
+            forced_leaves.push_back(std::move(item));
+        } else {
+            normal_leaves.push_back(std::move(item));
+        }
+    }
+
+    // Apply _column_paths_max quota to force leaves (sort by hits desc for determinism).
+    std::sort(forced_leaves.begin(), forced_leaves.end(),
+              [](const auto& a, const auto& b) { return a.first->hits > b.first->hits; });
+    size_t force_limit = (_column_paths_max > 0)
+                                 ? static_cast<size_t>(_column_paths_max)
+                                 : std::numeric_limits<size_t>::max();
+    for (size_t i = force_limit; i < forced_leaves.size(); i++) {
+        forced_leaves[i].first->remain = true;
+        _has_remain = true;
+    }
+    if (forced_leaves.size() > force_limit) {
+        forced_leaves.resize(force_limit);
+    }
+
+    // Apply _max_column quota to normal leaves (existing behaviour).
+    std::sort(normal_leaves.begin(), normal_leaves.end(),
+              [](const auto& a, const auto& b) { return a.first->hits > b.first->hits; });
+    size_t limit = _max_column > 0 ? static_cast<size_t>(_max_column) : std::numeric_limits<size_t>::max();
+    for (size_t i = limit; i < normal_leaves.size(); i++) {
+        if (!normal_leaves[i].first->remain && normal_leaves[i].first->hits >= _total_rows) {
             limit++;
             continue;
         }
-        hit_leaf[i].first->remain = true;
+        normal_leaves[i].first->remain = true;
     }
-    if (hit_leaf.size() > limit) {
+    if (normal_leaves.size() > limit) {
         _has_remain |= true;
-        hit_leaf.resize(limit);
+        normal_leaves.resize(limit);
     }
+
+    // Merge back and sort by path name for stable column order.
+    hit_leaf.clear();
+    hit_leaf.insert(hit_leaf.end(), forced_leaves.begin(), forced_leaves.end());
+    hit_leaf.insert(hit_leaf.end(), normal_leaves.begin(), normal_leaves.end());
     std::sort(hit_leaf.begin(), hit_leaf.end(), [](const auto& a, const auto& b) { return a.second < b.second; });
     for (auto& [node, path] : hit_leaf) {
         node->index = _paths.size();