fix metadata reload and add dict_get tests

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/scalar/DictGet.java

@@ -86,8 +86,12 @@ public Pair<FunctionSignature, Dictionary> customSignatureDict() {
         Dictionary dictionary;
         try {
             dictionary = dicMgr.getDictionary(dbName, dictName);
+            // check is not key column
+            if (dictionary.getDicColumns().stream().anyMatch(col -> col.getName().equals(colName) && col.isKey())) {
+                throw new AnalysisException("Can't ask for key " + colName + " by dict_get()");
+            }
         } catch (DdlException e) {
-            throw new AnalysisException("Dictionary " + dictName + " not found in database " + dbName);
+            throw new AnalysisException(e.getMessage());
         }
 
         return Pair.of(FunctionSignature.ret(dictionary.getColumnType(colName))

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/info/DictionaryColumnDefinition.java

@@ -18,15 +18,21 @@
 package org.apache.doris.nereids.trees.plans.commands.info;
 
 import org.apache.doris.catalog.Type;
+import org.apache.doris.common.io.Text;
+import org.apache.doris.common.io.Writable;
+import org.apache.doris.persist.gson.GsonUtils;
 
-import com.aliyun.datalake20200710.external.google.gson.annotations.SerializedName;
+import com.google.gson.annotations.SerializedName;
 
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
 import java.util.Objects;
 
 /**
  * Definition of a dictionary column.
  */
-public class DictionaryColumnDefinition {
+public class DictionaryColumnDefinition implements Writable {
     @SerializedName(value = "name")
     private final String name;
 
@@ -41,6 +47,12 @@ public DictionaryColumnDefinition(String name, boolean isKey) {
         this.isKey = isKey;
     }
 
+    public DictionaryColumnDefinition(String name, boolean isKey, Type type) {
+        this.name = Objects.requireNonNull(name, "Column name cannot be null");
+        this.isKey = isKey;
+        this.type = type;
+    }
+
     public String getName() {
         return name;
     }
@@ -56,4 +68,15 @@ public Type getType() {
     public void setType(Type type) {
         this.type = type;
     }
+
+    @Override
+    public void write(DataOutput out) throws IOException {
+        String json = GsonUtils.GSON.toJson(this);
+        Text.writeString(out, json);
+    }
+
+    public static DictionaryColumnDefinition read(DataInput in) throws IOException {
+        String json = Text.readString(in);
+        return GsonUtils.GSON.fromJson(json, DictionaryColumnDefinition.class);
+    }
 }

regression-test/suites/dictionary_p0/test_dict_get.groovy

@@ -0,0 +1,82 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_dict_get") {
+    sql "drop database if exists test_dictionary_function"
+    sql "create database test_dictionary_function"
+    sql "use test_dictionary_function"
+
+    sql """
+        create table dc(
+            k0 datetime(6) not null,
+            k1 varchar not null
+        )
+        DISTRIBUTED BY HASH(`k0`) BUCKETS auto
+        properties("replication_num" = "1");
+    """
+    sql """insert into dc values('2020-12-12', 'abc');"""
+
+    sql """
+        create dictionary dic1 using dc
+        (
+            k0 KEY,
+            k1 VALUE
+        )
+        LAYOUT(HASH_MAP);
+    """
+    explain {
+        sql """select dict_get("test_dictionary_function.dic1", "k1", "2020-12-12") from dc"""
+        verbose true
+        contains "type=datetimev2(6)"
+        contains "type=varchar(65533)"
+    }
+    test {
+        sql """select dict_get("test_dictionary_function.dic1", "k0", "2020-12-12") from dc"""
+        exception "Can't ask for key k0 by dict_get()"
+    }
+
+    sql """
+        create dictionary dic2 using dc
+        (
+            k1 KEY,
+            k0 VALUE
+        )
+        LAYOUT(HASH_MAP);
+    """
+    explain {
+        sql """select dict_get("test_dictionary_function.dic2", "k0", "abc") from dc"""
+        verbose true
+        contains "type=datetimev2(6)"
+        notContains "type=varchar(65533)"
+    }
+
+
+    sql """
+        create dictionary dic_ip_trie using dc
+        (
+            k1 KEY,
+            k0 VALUE
+        )
+        LAYOUT(IP_TRIE);
+    """
+    explain {
+        sql """select dict_get("test_dictionary_function.dic_ip_trie", "k0", "abc") from dc"""
+        verbose true
+        contains "type=datetimev2(6)"
+        notContains "type=varchar(65533)"
+    }
+}