fix(tianmu):Resolve DDL and insert inot select Possible space inflati…

…on issues(stoneatom#366)

Cause of the problem:
1. For multiple versions, Tianmu needs to first copy an original pack when performing DML operations,
Modify the copied package and use append write or overwrite write after modification
(If there is invalid space in the DATA file that can be written to the current pack, use overwrite write, otherwise use append write) to write to the file,
After the latest package is written to a file, the latest version chain will point to the address that was last written.
There is a problem with the current (TianmuAttr:: LoadData) logic. Every time you call (TianmuAttr:: LoadData),
Will write data to disk,
If there are multiple rows written in a transaction, there will be multiple copies of data,
"Because the current transaction has not been committed, the space for previous repeated writes has not been released, so the logic of overwriting writes will not be reached.",
"I only follow the logic of additional writing, which is the fundamental reason for the skyrocketing space.".
If you encounter a particularly large multiline write transaction, it will lead to a space explosion.
Moreover, disk IO is performed once per load line, which can also lead to degraded insert performance.
Solution:
To optimize the logic of (TianmuAttr:: LoadData), it is necessary to determine whether the data in the pack is full before saving changes,
Is whether to reach 65536 lines, and if so, write again,
If it cannot be reached, it is necessary to write again in the commit phase.

storage/tianmu/core/tianmu_attr.cpp

@@ -905,13 +905,12 @@ void TianmuAttr::LoadData(loader::ValueCache *nvs, Transaction *conn_info) {
       throw common::DatabaseException("Unknown pack type" + Path().string());
       break;
   }
-
   DPN &dpn = get_dpn(pi);
-  Pack *pack = get_pack(pi);
-  if (!dpn.Trivial()) {
-    pack->Save();
-  }
-  if (current_txn_->LoadSource() == common::LoadSource::LS_File) {
+  if (current_txn_->LoadSource() == common::LoadSource::LS_File || dpn.numOfRecords == (1U << pss)) {
+    Pack *pack = get_pack(pi);
+    if (!dpn.Trivial()) {
+      pack->Save();
+    }
     if (pack) {
       pack->Unlock();
     }