[fix](stream_load) fix stream load may failed caused by column name with keyword (#35822) (#37890)

#35938 #35822 
let 
KW_SQL,
KW_CACHE,
KW_COLOCATE,
KW_COMPRESS_TYPE,
KW_DORIS_INTERNAL_TABLE_ID,
KW_HOTSPOT,
KW_PRIVILEGES,
KW_RECENT,
KW_STAGES,
KW_WARM,
KW_UP,
KW_CONVERT_LSC
be as non-reserved

## Proposed changes

Issue Number: close #xxx

<!--Describe your changes.-->

---------

Co-authored-by: caiconghui1 <caiconghui1@jd.com>

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisLexer.g4

@@ -139,6 +139,7 @@ BUILD: 'BUILD';
 BUILTIN: 'BUILTIN';
 BULK: 'BULK';
 BY: 'BY';
+CACHE: 'CACHE';
 CACHED: 'CACHED';
 CALL: 'CALL';
 CANCEL: 'CANCEL';
@@ -156,6 +157,7 @@ CLUSTERS: 'CLUSTERS';
 COLLATE: 'COLLATE';
 COLLATION: 'COLLATION';
 COLLECT: 'COLLECT';
+COLOCATE: 'COLOCATE';
 COLUMN: 'COLUMN';
 COLUMNS: 'COLUMNS';
 COMMENT: 'COMMENT';
@@ -163,6 +165,7 @@ COMMIT: 'COMMIT';
 COMMITTED: 'COMMITTED';
 COMPACT: 'COMPACT';
 COMPLETE: 'COMPLETE';
+COMPRESS_TYPE: 'COMPRESS_TYPE';
 CONFIG: 'CONFIG';
 CONNECTION: 'CONNECTION';
 CONNECTION_ID: 'CONNECTION_ID';
@@ -170,6 +173,7 @@ CONSISTENT: 'CONSISTENT';
 CONSTRAINT: 'CONSTRAINT';
 CONSTRAINTS: 'CONSTRAINTS';
 CONVERT: 'CONVERT';
+CONVERT_LSC: 'CONVERT_LSC';
 COPY: 'COPY';
 COUNT: 'COUNT';
 CREATE: 'CREATE';
@@ -288,6 +292,7 @@ HISTOGRAM: 'HISTOGRAM';
 HLL: 'HLL';
 HLL_UNION: 'HLL_UNION';
 HOSTNAME: 'HOSTNAME';
+HOTSPOT: 'HOTSPOT';
 HOUR: 'HOUR';
 HUB: 'HUB';
 IDENTIFIED: 'IDENTIFIED';
@@ -410,6 +415,7 @@ PERMISSIVE: 'PERMISSIVE';
 PHYSICAL: 'PHYSICAL';
 PLACEHOLDER: '?';
 PLAN: 'PLAN';
+PRIVILEGES: 'PRIVILEGES';
 PROCESS: 'PROCESS';
 PLUGIN: 'PLUGIN';
 PLUGINS: 'PLUGINS';
@@ -432,6 +438,7 @@ RANGE: 'RANGE';
 READ: 'READ';
 REAL: 'REAL';
 REBALANCE: 'REBALANCE';
+RECENT: 'RECENT';
 RECOVER: 'RECOVER';
 RECYCLE: 'RECYCLE';
 REFRESH: 'REFRESH';
@@ -485,7 +492,9 @@ SMALLINT: 'SMALLINT';
 SNAPSHOT: 'SNAPSHOT';
 SONAME: 'SONAME';
 SPLIT: 'SPLIT';
+SQL: 'SQL';
 SQL_BLOCK_RULE: 'SQL_BLOCK_RULE';
+STAGES: 'STAGES';
 START: 'START';
 STARTS: 'STARTS';
 STATS: 'STATS';
@@ -537,6 +546,7 @@ UNION: 'UNION';
 UNIQUE: 'UNIQUE';
 UNLOCK: 'UNLOCK';
 UNSIGNED: 'UNSIGNED';
+UP: 'UP';
 UPDATE: 'UPDATE';
 USE: 'USE';
 USER: 'USER';
@@ -549,6 +559,7 @@ VARIANT: 'VARIANT';
 VERBOSE: 'VERBOSE';
 VERSION: 'VERSION';
 VIEW: 'VIEW';
+WARM: 'WARM';
 WARNINGS: 'WARNINGS';
 WEEK: 'WEEK';
 WHEN: 'WHEN';

fe/fe-core/src/main/antlr4/org/apache/doris/nereids/DorisParser.g4

@@ -1047,6 +1047,7 @@ nonReserved
     | BUILD
     | BUILTIN
     | BULK
+    | CACHE
     | CACHED
     | CALL
     | CATALOG
@@ -1059,18 +1060,21 @@ nonReserved
     | CLUSTERS
     | COLLATION
     | COLLECT
+    | COLOCATE
     | COLUMNS
     | COMMENT
     | COMMIT
     | COMMITTED
     | COMPACT
     | COMPLETE
+    | COMPRESS_TYPE
     | CONFIG
     | CONNECTION
     | CONNECTION_ID
     | CONSISTENT
     | CONSTRAINTS
     | CONVERT
+    | CONVERT_LSC
     | COPY
     | COUNT
     | CREATION
@@ -1144,6 +1148,7 @@ nonReserved
     | HISTOGRAM
     | HLL_UNION
     | HOSTNAME
+    | HOTSPOT
     | HOUR
     | HUB
     | IDENTIFIED
@@ -1233,6 +1238,7 @@ nonReserved
     | PLUGIN
     | PLUGINS
     | POLICY
+    | PRIVILEGES
     | PROC
     | PROCESS
     | PROCESSLIST
@@ -1244,6 +1250,7 @@ nonReserved
     | QUERY
     | QUOTA
     | RANDOM
+    | RECENT
     | RECOVER
     | RECYCLE
     | REFRESH
@@ -1278,6 +1285,8 @@ nonReserved
     | SNAPSHOT
     | SONAME
     | SPLIT
+    | SQL
+    | STAGES
     | START
     | STARTS
     | STATS
@@ -1308,6 +1317,7 @@ nonReserved
     | TYPES
     | UNCOMMITTED
     | UNLOCK
+    | UP
     | USER
     | VALUE
     | VARCHAR
@@ -1316,6 +1326,7 @@ nonReserved
     | VERBOSE
     | VERSION
     | VIEW
+    | WARM
     | WARNINGS
     | WEEK
     | WORK

fe/fe-core/src/main/cup/sql_parser.cup

@@ -287,6 +287,7 @@ terminal String
     KW_BUILD,
     KW_BUILTIN,
     KW_BY,
+    KW_CACHE,
     KW_CACHED,
     KW_CANCEL,
     KW_CASE,
@@ -418,6 +419,7 @@ terminal String
     KW_HELP,
     KW_HLL,
     KW_HLL_UNION,
+    KW_HOTSPOT,
     KW_HOSTNAME,
     KW_HOUR,
     KW_HUB,
@@ -597,6 +599,7 @@ terminal String
     KW_SPLIT,
     KW_SQL,
     KW_SQL_BLOCK_RULE,
+    KW_STAGES,
     KW_START,
     KW_STARTS,    
     KW_STATS,
@@ -645,6 +648,7 @@ terminal String
     KW_UNLOCK,
     KW_UNSET,
     KW_UNSIGNED,
+    KW_UP,
     KW_UPDATE,
     KW_USE,
     KW_USER,
@@ -659,6 +663,7 @@ terminal String
     KW_VIEW,
     KW_VIEWS,
     KW_WARNINGS,
+    KW_WARM,
     KW_WEEK,
     KW_WHEN,
     KW_WHERE,
@@ -8192,6 +8197,30 @@ keyword ::=
     {: RESULT = id; :}
     | KW_MATCH_PHRASE_EDGE:id
     {: RESULT = id; :}
+    | KW_SQL:id
+    {: RESULT = id; :}
+    | KW_CACHE:id
+    {: RESULT = id; :}
+    | KW_COLOCATE:id
+    {: RESULT = id; :}
+    | KW_COMPRESS_TYPE:id
+    {: RESULT = id; :}
+    | KW_DORIS_INTERNAL_TABLE_ID:id
+    {: RESULT = id; :}
+    | KW_HOTSPOT:id
+    {: RESULT = id; :}
+    | KW_PRIVILEGES:id
+    {: RESULT = id; :}
+    | KW_RECENT:id
+    {: RESULT = id; :}
+    | KW_STAGES:id
+    {: RESULT = id; :}
+    | KW_WARM:id
+    {: RESULT = id; :}
+    | KW_UP:id
+    {: RESULT = id; :}
+    | KW_CONVERT_LSC:id
+    {: RESULT = id; :}
     ;
 
 // Identifier that contain keyword

regression-test/data/load_p0/stream_load/test_keyword.csv

@@ -0,0 +1,10 @@
+1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16
+2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17
+3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18
+4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19
+5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20
+6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21
+7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22
+8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23
+9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24
+10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25

regression-test/data/load_p0/stream_load/test_stream_load_keyword.out

@@ -0,0 +1,13 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+1	2	3	4	5	6	7	8	9	10	11	12	13	14	15	16
+2	3	4	5	6	7	8	9	10	11	12	13	14	15	16	17
+3	4	5	6	7	8	9	10	11	12	13	14	15	16	17	18
+4	5	6	7	8	9	10	11	12	13	14	15	16	17	18	19
+5	6	7	8	9	10	11	12	13	14	15	16	17	18	19	20
+6	7	8	9	10	11	12	13	14	15	16	17	18	19	20	21
+7	8	9	10	11	12	13	14	15	16	17	18	19	20	21	22
+8	9	10	11	12	13	14	15	16	17	18	19	20	21	22	23
+9	10	11	12	13	14	15	16	17	18	19	20	21	22	23	24
+10	11	12	13	14	15	16	17	18	19	20	21	22	23	24	25
+

regression-test/suites/load_p0/stream_load/test_stream_load_keyword.groovy

@@ -0,0 +1,75 @@
+// 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_stream_load_keyword", "p0") {
+    def tableName = "test_stream_load_keyword"
+    try {
+        sql """ DROP TABLE IF EXISTS ${tableName} """
+        sql """
+            CREATE TABLE IF NOT EXISTS ${tableName} (
+                `k1` bigint(20) NULL,
+                `k2` bigint(20) NULL,
+                 sql int(11) SUM NULL,
+                 cache int(11) REPLACE NULL,
+                 colocate int(11) REPLACE_IF_NOT_NULL NULL,
+                 compress_type int(11) REPLACE_IF_NOT_NULL NULL,
+                 doris_internal_table_id int(11) REPLACE_IF_NOT_NULL NULL,
+                 `dual` int(11) REPLACE_IF_NOT_NULL NULL,
+                 hotspot int(11) REPLACE_IF_NOT_NULL NULL,
+                 `overwrite` int(11) REPLACE_IF_NOT_NULL NULL,
+                 privileges int(11) REPLACE_IF_NOT_NULL NULL,
+                 recent int(11) REPLACE_IF_NOT_NULL NULL,
+                 stages int(11) REPLACE_IF_NOT_NULL NULL,
+                 warm int(11) REPLACE_IF_NOT_NULL NULL,
+                 up int(11) REPLACE_IF_NOT_NULL NULL,
+                 convert_lsc int(11) REPLACE_IF_NOT_NULL NULL,
+            ) ENGINE=OLAP
+            AGGREGATE KEY(`k1`, `k2`)
+            COMMENT 'OLAP'
+            DISTRIBUTED BY HASH(`k1`, `k2`) BUCKETS 3
+            PROPERTIES ("replication_allocation" = "tag.location.default: 1");
+        """
+
+        // test columns with keyword success
+        streamLoad {
+            table "${tableName}"
+
+            file 'test_keyword.csv'
+
+            set 'columns', 'k1, k2, sql, cache, colocate, compress_type, doris_internal_table_id, `dual`,' +
+                    ' hotspot, `overwrite`, privileges, recent, stages, warm, up, convert_lsc'
+
+            set 'column_separator', ','
+
+            check { result, exception, startTime, endTime ->
+                if (exception != null) {
+                    throw exception
+                }
+                log.info("Stream load result: ${result}".toString())
+                def json = parseJson(result)
+                assertEquals("success", json.Status.toLowerCase())
+                assertEquals(10, json.NumberTotalRows)
+            }
+            time 10000 // limit inflight 10s
+        }
+
+        sql "sync"
+        qt_sql "select * from ${tableName} order by k1"
+    } finally {
+        sql """ DROP TABLE IF EXISTS ${tableName} """
+    }
+}