[pick](variant) pick some serde impl (#40573)

be/src/vec/data_types/serde/data_type_nothing_serde.h

@@ -107,6 +107,13 @@ class DataTypeNothingSerde : public DataTypeSerDe {
                                std::vector<StringRef>& buffer_list) const override {
         return Status::NotSupported("write_column_to_orc with type " + column.get_name());
     }
+
+    Status write_one_cell_to_json(const IColumn& column, rapidjson::Value& result,
+                                  rapidjson::Document::AllocatorType& allocator, Arena& mem_pool,
+                                  int row_num) const override {
+        result.SetNull();
+        return Status::OK();
+    }
 };
 } // namespace vectorized
 } // namespace doris

be/src/vec/data_types/serde/data_type_object_serde.cpp

@@ -160,6 +160,37 @@ void DataTypeObjectSerDe::write_column_to_arrow(const IColumn& column, const Nul
     }
 }
 
+Status DataTypeObjectSerDe::write_one_cell_to_json(const IColumn& column, rapidjson::Value& result,
+                                                   rapidjson::Document::AllocatorType& allocator,
+                                                   Arena& mem_pool, int row_num) const {
+    const auto& var = assert_cast<const ColumnObject&>(column);
+    if (!var.is_finalized()) {
+        var.assume_mutable()->finalize();
+    }
+    result.SetObject();
+    // sort to make output stable, todo add a config
+    auto subcolumns = schema_util::get_sorted_subcolumns(var.get_subcolumns());
+    for (const auto& entry : subcolumns) {
+        const auto& subcolumn = entry->data.get_finalized_column();
+        const auto& subtype_serde = entry->data.get_least_common_type_serde();
+        if (subcolumn.is_null_at(row_num)) {
+            continue;
+        }
+        rapidjson::Value key;
+        key.SetString(entry->path.get_path().data(), entry->path.get_path().size());
+        rapidjson::Value val;
+        RETURN_IF_ERROR(subtype_serde->write_one_cell_to_json(subcolumn, val, allocator, mem_pool,
+                                                              row_num));
+        if (val.IsNull() && entry->path.empty()) {
+            // skip null value with empty key, indicate the null json value of root in variant map,
+            // usally padding in nested arrays
+            continue;
+        }
+        result.AddMember(key, val, allocator);
+    }
+    return Status::OK();
+}
+
 } // namespace vectorized
 
 } // namespace doris

be/src/vec/data_types/serde/data_type_object_serde.h

@@ -93,6 +93,10 @@ class DataTypeObjectSerDe : public DataTypeSerDe {
         return Status::NotSupported("write_column_to_orc with type " + column.get_name());
     }
 
+    Status write_one_cell_to_json(const IColumn& column, rapidjson::Value& result,
+                                  rapidjson::Document::AllocatorType& allocator, Arena& mem_pool,
+                                  int row_num) const override;
+
 private:
     template <bool is_binary_format>
     Status _write_column_to_mysql(const IColumn& column, MysqlRowBuffer<is_binary_format>& result,