[branch-1.2](es catalog) fix issue with select and insert from es cat…

…alog core (#39585)

## Proposed changes

Issue Number: pick #24318

```
(gdb) 
#0  0x00007fb77df2b387 in raise () from /lib64/libc.so.6
#1  0x00007fb77df2ca78 in abort () from /lib64/libc.so.6
#2  0x000056189ef8e1d9 in ?? ()
#3  0x000056189ef837ed in google::LogMessage::Fail() ()
#4  0x000056189ef85d29 in google::LogMessage::SendToLog() ()
#5  0x000056189ef83356 in google::LogMessage::Flush() ()
#6  0x000056189ef86399 in google::LogMessageFatal::~LogMessageFatal() ()
#7  0x000056189ac09e08 in doris::vectorized::IColumn::append_data_by_selector_impl<doris::vectorized::ColumnNullable> (selector=..., res=..., 
    this=<optimized out>) at /var/local/ldb-toolchain/include/c++/11/ext/new_allocator.h:89
#8  doris::vectorized::ColumnNullable::append_data_by_selector (this=<optimized out>, res=..., selector=...)
    at /data/TCHouse-D-1.2/be/src/vec/columns/column_nullable.h:201
#9  0x000056189acab325 in doris::vectorized::Block::append_block_by_selector (this=this@entry=0x7fb6a9e3a510, columns=..., selector=...)
    at /var/local/ldb-toolchain/include/c++/11/bits/stl_vector.h:1043
#10 0x000056189e90e9a8 in doris::stream_load::VNodeChannel::add_block (this=0x56198399b090, block=0x7fb6a9e3a510, payload=...)
    at /data/TCHouse-D-1.2/be/src/vec/core/block.h:421
#11 0x000056189e914840 in doris::stream_load::VOlapTableSink::send (this=<optimized out>, state=<optimized out>, input_block=<optimized out>)
    at /data/TCHouse-D-1.2/be/src/vec/sink/vtablet_sink.cpp:608
#12 0x0000561899b13261 in doris::PlanFragmentExecutor::open_vectorized_internal (this=this@entry=0x56199fdaaf80)
    at /data/TCHouse-D-1.2/be/src/runtime/plan_fragment_executor.cpp:322
#13 0x0000561899b1423e in doris::PlanFragmentExecutor::open (this=this@entry=0x56199fdaaf80)
    at /data/TCHouse-D-1.2/be/src/runtime/plan_fragment_executor.cpp:261
#14 0x0000561899aec9a0 in doris::FragmentExecState::execute (this=0x56199fdaaf00) at /data/TCHouse-D-1.2/be/src/runtime/fragment_mgr.cpp:260
#15 0x0000561899aeffae in doris::FragmentMgr::_exec_actual(std::shared_ptr<doris::FragmentExecState>, std::function<void (doris::PlanFragmentExecutor*)>) (this=this@entry=0x5618a9b11400, exec_state=..., cb=...) at /var/local/ldb-toolchain/include/c++/11/bits/shared_ptr_base.h:1290
#16 0x0000561899af04e2 in operator() (__closure=<optimized out>) at /data/TCHouse-D-1.2/be/src/runtime/fragment_mgr.cpp:737
#17 std::__invoke_impl<void, doris::FragmentMgr::exec_plan_fragment(const doris::TExecPlanFragmentParams&, doris::FragmentMgr::FinishCallback)::<lambda()>&> (__f=...) at /var/local/ldb-toolchain/include/c++/11/bits/invoke.h:61
#18 std::__invoke_r<void, doris::FragmentMgr::exec_plan_fragment(const doris::TExecPlanFragmentParams&, doris::FragmentMgr::FinishCallback)::<lambda()>&> (__fn=...) at /var/local/ldb-toolchain/include/c++/11/bits/invoke.h:111
#19 std::_Function_handler<void(), doris::FragmentMgr::exec_plan_fragment(const doris::TExecPlanFragmentParams&, doris::FragmentMgr::FinishCallback)::<lambda()> >::_M_invoke(const std::_Any_data &) (__functor=...) at /var/local/ldb-toolchain/include/c++/11/bits/std_function.h:291
#20 0x0000561899d9fcb5 in std::function<void ()>::operator()() const (this=<optimized out>)
    at /var/local/ldb-toolchain/include/c++/11/bits/std_function.h:560
#21 doris::FunctionRunnable::run (this=<optimized out>) at /data/TCHouse-D-1.2/be/src/util/threadpool.cpp:46
#22 doris::ThreadPool::dispatch_thread (this=0x5618ab4e0700) at /data/TCHouse-D-1.2/be/src/util/threadpool.cpp:535
#23 0x0000561899d9510f in std::function<void ()>::operator()() const (this=0x5618a9e02ef8)
    at /var/local/ldb-toolchain/include/c++/11/bits/std_function.h:560
#24 doris::Thread::supervise_thread (arg=0x5618a9e02ee0) at /data/TCHouse-D-1.2/be/src/util/thread.cpp:454
#25 0x00007fb77dce0ea5 in start_thread () from /lib64/libpthread.so.0
#26 0x00007fb77dff39fd in clone () from /lib64/libc.so.6
```

<!--Describe your changes.-->

be/src/exec/es/es_scroll_parser.cpp

@@ -334,7 +334,16 @@ template <typename T>
 static Status insert_int_value(const rapidjson::Value& col, PrimitiveType type,
                                vectorized::IColumn* col_ptr, bool pure_doc_value, bool nullable) {
     if (col.IsNumber()) {
-        T value = (T)(sizeof(T) < 8 ? col.GetInt() : col.GetInt64());
+        T value;
+        // ES allows inserting float and double in int/long types.
+        // To parse these numbers in Doris, we direct cast them to int types.
+        if (col.IsDouble()) {
+            value = static_cast<T>(col.GetDouble());
+        } else if (col.IsFloat()) {
+            value = static_cast<T>(col.GetFloat());
+        } else {
+            value = (T)(sizeof(T) < 8 ? col.GetInt() : col.GetInt64());
+        }
         col_ptr->insert_data(const_cast<const char*>(reinterpret_cast<char*>(&value)), 0);
         return Status::OK();
     }
@@ -350,8 +359,14 @@ static Status insert_int_value(const rapidjson::Value& col, PrimitiveType type,
     RETURN_ERROR_IF_COL_IS_NOT_STRING(col, type);
 
     StringParser::ParseResult result;
-    const std::string& val = col.GetString();
-    size_t len = col.GetStringLength();
+    std::string val = col.GetString();
+    // ES allows inserting numbers and characters containing decimals in numeric types.
+    // To parse these numbers in Doris, we remove the decimals here.
+    size_t pos = val.find(".");
+    if (pos != std::string::npos) {
+        val = val.substr(0, pos);
+    }
+    size_t len = val.length();
     T v = StringParser::string_to_int<T>(val.c_str(), len, &result);
     RETURN_ERROR_IF_PARSING_FAILED(result, col, type);
 

docker/thirdparties/docker-compose/elasticsearch/scripts/data/data1.json

@@ -7,6 +7,7 @@
   "test6": 1659931810000,
   "test7": 1659931810000,
   "test8": "2022-08-08T12:10:10Z",
+  "test9": 12345,
   "c_bool": [true, false, true, true],
   "c_byte": [1, -2, -3, 4],
   "c_short": [128, 129, -129, -130],
@@ -26,4 +27,4 @@
     {"name": "Andy", "age": 18},
     {"name": "Tim", "age": 28}
   ]
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/data/data1_es6.json

@@ -3,6 +3,7 @@
   "test2": "text#1",
   "test3": 3.14,
   "test4": "2022-08-08",
+  "test5": 12345,
   "c_bool": [true, false, true, true],
   "c_byte": [1, -2, -3, 4],
   "c_short": [128, 129, -129, -130],
@@ -22,4 +23,4 @@
     {"name": "Andy", "age": 18},
     {"name": "Tim", "age": 28}
   ]
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/data/data2.json

@@ -7,6 +7,7 @@
   "test6": 1660018210000,
   "test7": "2022-08-09 12:10:10",
   "test8": 1660018210000,
+  "test9": 2222.2,
   "c_bool": [true, false, true, true],
   "c_byte": [1, -2, -3, 4],
   "c_short": [128, 129, -129, -130],
@@ -26,4 +27,4 @@
     {"name": "Andy", "age": 18},
     {"name": "Tim", "age": 28}
   ]
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/data/data2_es6.json

@@ -3,6 +3,7 @@
   "test2": "text2",
   "test3": 4,
   "test4": "2022-08-08",
+  "test5": 2222.2,
   "c_bool": [true, false, true, true],
   "c_byte": [1, -2, -3, 4],
   "c_short": [128, 129, -129, -130],
@@ -22,4 +23,4 @@
     {"name": "Andy", "age": 18},
     {"name": "Tim", "age": 28}
   ]
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/data/data3.json

@@ -7,6 +7,7 @@
   "test6": 1660104610000,
   "test7": 1660104610000,
   "test8": "2022-08-10T12:10:10",
+  "test9": 3333.22,
   "c_bool": [true, false, true, true],
   "c_byte": [1, -2, -3, 4],
   "c_short": [128, 129, -129, -130],
@@ -26,4 +27,4 @@
     {"name": "Andy", "age": 18},
     {"name": "Tim", "age": 28}
   ]
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/data/data3_es6.json

@@ -3,6 +3,7 @@
   "test2": "text3_4*5",
   "test3": 5.0,
   "test4": "2022-08-08",
+  "test5": "3333.22",
   "c_bool": [true, false, true, true],
   "c_byte": [1, -2, -3, 4],
   "c_short": [128, 129, -129, -130],
@@ -22,4 +23,4 @@
     {"name": "Andy", "age": 18},
     {"name": "Tim", "age": 28}
   ]
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/data/data4.json

@@ -7,6 +7,7 @@
   "test6": 1660191010000,
   "test7": "2022-08-11 12:10:10",
   "test8": "2022-08-11T12:10:10+09:00",
+  "test9": "4444.22",
   "c_bool": [true, false, true, true],
   "c_byte": [1, -2, -3, 4],
   "c_short": [128, 129, -129, -130],
@@ -26,4 +27,4 @@
     {"name": "Andy", "age": 18},
     {"name": "Tim", "age": 28}
   ]
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/index/es6_test1.json

@@ -24,6 +24,9 @@
         "test4": {
           "type": "date"
         },
+        "test5": {
+          "type": "long"
+        },
         "c_bool": {
           "type": "boolean"
         },
@@ -85,4 +88,4 @@
       }
     }
   }
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/index/es6_test2.json

@@ -27,6 +27,9 @@
         "test4": {
           "type": "date"
         },
+        "test5": {
+          "type": "long"
+        },
         "c_bool": {
           "type": "boolean"
         },
@@ -88,4 +91,4 @@
       }
     }
   }
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/index/es7_test1.json

@@ -38,6 +38,9 @@
       "test8": {
         "type": "date"
       },
+      "test9": {
+        "type": "long"
+      },
       "c_bool": {
         "type": "boolean"
       },
@@ -98,4 +101,4 @@
       }
     }
   }
-}
+}

docker/thirdparties/docker-compose/elasticsearch/scripts/index/es7_test2.json

@@ -41,6 +41,9 @@
       "test8": {
         "type": "date"
       },
+      "test9": {
+        "type": "long"
+      },
       "c_bool": {
         "type": "boolean"
       },
@@ -101,4 +104,4 @@
       }
     }
   }
-}
+}