Merge remote-tracking branch 'upstream/main' into mmap2

CMakeLists.txt

@@ -68,7 +68,7 @@ execute_process(
         OUTPUT_VARIABLE GIT_BRANCH_NAME
         OUTPUT_STRIP_TRAILING_WHITESPACE)
 execute_process(
-        COMMAND "${GIT_EXECUTABLE}" rev-parse HEAD
+        COMMAND "${GIT_EXECUTABLE}" describe --tags --match=v* --first-parent --always
         OUTPUT_VARIABLE GIT_COMMIT_ID
         OUTPUT_STRIP_TRAILING_WHITESPACE)
 if("${GIT_BRANCH_NAME}" STREQUAL "")
@@ -260,7 +260,7 @@ if(DEFINED CPACK_PACKAGE_VERSION)
     string(REPLACE "-" "." CPACK_PACKAGE_VERSION "${CPACK_PACKAGE_VERSION}")
 endif()
 if(NOT DEFINED CPACK_PACKAGE_VERSION OR CPACK_PACKAGE_VERSION STREQUAL "")
-    set(CPACK_PACKAGE_VERSION "0.5.2")
+    set(CPACK_PACKAGE_VERSION "${GIT_COMMIT_ID}")
 endif()
 set(CPACK_PACKAGE_RELEASE 1)
 set(CPACK_PACKAGE_CONTACT "Zhichang Yu <[email protected]>")

README.md

@@ -82,6 +82,8 @@ Infinity supports two working modes, embedded mode and client-server mode. Infin
    print(res)
    ```
 
+> 💡 For more information about Infinity's Python API, see the [Python API Reference](https://infiniflow.org/docs/dev/pysdk_api_reference).
+
 #### 🔧 Deploy Infinity in client-server mode
 
 If you wish to deploy Infinity with the server and client as separate processes, see the [Deploy infinity server](https://infiniflow.org/docs/dev/deploy_infinity_server) guide.
@@ -90,8 +92,6 @@ If you wish to deploy Infinity with the server and client as separate processes,
 
 See the [Build from Source](https://infiniflow.org/docs/dev/build_from_source) guide.
 
-> 💡 For more information about Infinity's Python API, see the [Python API Reference](https://infiniflow.org/docs/dev/pysdk_api_reference).
-
 ## 📚 Document
 
 - [Quickstart](https://infiniflow.org/docs/dev/)

conf/infinity_conf.toml

@@ -43,6 +43,8 @@ mem_index_capacity       = 65536
 # secret_key               = "minioadmin"
 # enable_https             = false
 
+snapshot_dir            = "/var/infinity/snapshots"
+
 [buffer]
 buffer_manager_size      = "4GB"
 lru_num                  = 7

python/test_pysdk/test_knn.py

@@ -1069,7 +1069,7 @@ def test_sparse_knn_with_index(self, check_data, suffix):
                                                index.IndexType.BMP,
                                                {"block_size": "8", "compress_type": "compress"}), ConflictType.Error)
 
-        table_obj.optimize("idx1", {"topk": "3"})
+        # table_obj.optimize("idx1", {"topk": "3"})
 
         res, extra_result = (table_obj
                              .output(["*", "_row_id", "_similarity"])
@@ -1592,7 +1592,7 @@ def test_sparse_knn_with_invalid_alpha_beta(self, check_data, alpha, beta, suffi
                                                index.IndexType.BMP,
                                                {"block_size": "8", "compress_type": "compress"}), ConflictType.Error)
 
-        table_obj.optimize("idx1", {"topk": "3"})
+        # table_obj.optimize("idx1", {"topk": "3"})
 
         with pytest.raises(InfinityException) as e:
             res, extra_result = (table_obj
@@ -1910,7 +1910,7 @@ def test_match_sparse_index_hint(self, check_data, suffix):
                                                index.IndexType.BMP,
                                                {"block_size": "8", "compress_type": "raww"}), ConflictType.Error)
 
-        table_obj.optimize("idx1", {"topk": "3"})
+        # table_obj.optimize("idx1", {"topk": "3"})
 
         res, extra_result = (table_obj
                              .output(["*", "_row_id", "_similarity"])

src/admin/admin_executor.cpp

@@ -3399,6 +3399,28 @@ QueryResult AdminExecutor::ListConfigs(QueryContext *query_context, const AdminS
         }
     }
 
+    {
+        {
+            // option name
+            Value value = Value::MakeVarchar(SNAPSHOT_DIR_OPTION_NAME);
+            ValueExpression value_expr(value);
+            value_expr.AppendToChunk(output_block_ptr->column_vectors[0]);
+        }
+        {
+            // option name type
+            Value value = Value::MakeVarchar(std::to_string(global_config->CleanupInterval()));
+            ValueExpression value_expr(value);
+            value_expr.AppendToChunk(output_block_ptr->column_vectors[1]);
+        }
+        {
+            // option name type
+            Value value = Value::MakeVarchar("Snapshots store directory");
+            ValueExpression value_expr(value);
+            value_expr.AppendToChunk(output_block_ptr->column_vectors[2]);
+        }
+    }
+
+
     {
         {
             // option name
@@ -3483,6 +3505,27 @@ QueryResult AdminExecutor::ListConfigs(QueryContext *query_context, const AdminS
         }
     }
 
+    {
+        {
+            // option name
+            Value value = Value::MakeVarchar(SNAPSHOT_DIR_OPTION_NAME);
+            ValueExpression value_expr(value);
+            value_expr.AppendToChunk(output_block_ptr->column_vectors[0]);
+        }
+        {
+            // option name type
+            Value value = Value::MakeVarchar(global_config->SnapshotDir());
+            ValueExpression value_expr(value);
+            value_expr.AppendToChunk(output_block_ptr->column_vectors[1]);
+        }
+        {
+            // option name type
+            Value value = Value::MakeVarchar("Snapshot storage directory");
+            ValueExpression value_expr(value);
+            value_expr.AppendToChunk(output_block_ptr->column_vectors[2]);
+        }
+    }
+
     {
         {
             // option name

src/bin/infinity_main.cpp

@@ -162,6 +162,7 @@ void SignalHandler(int signal_number, siginfo_t *, void *) {
         }
         case SIGSEGV: {
             // Print back strace
+            infinity::PrintTransactionHistory();
             infinity::PrintStacktrace("SEGMENT FAULTS");
             exit(-1);
             break;
@@ -212,6 +213,7 @@ void TerminateHandler() {
     } catch (...) {
         message += "Unknown Unhandled Exception";
     }
+    infinity::PrintTransactionHistory();
     infinity::PrintStacktrace(message);
     std::abort();
 }

src/common/default_values.cppm

@@ -140,6 +140,8 @@ export {
     constexpr std::string_view SYSTEM_CONFIG_TABLE_NAME = "config";
     constexpr SizeT DEFAULT_PROFILER_HISTORY_SIZE = 128;
 
+    constexpr SizeT DEFAULT_TXN_HISTORY_SIZE = 128;
+
     // default emvb parameter
     constexpr u32 EMVB_CENTROID_NPROBE = 3;
     constexpr f32 EMVB_THRESHOLD_FIRST = 0.0f;
@@ -200,8 +202,10 @@ export {
     constexpr std::string_view DEFAULT_RESULT_CACHE = "off";
     constexpr SizeT DEFAULT_CACHE_RESULT_CAPACITY = 10000;
 
+    constexpr std::string_view DEFAULT_SNAPSHOT_DIR = "/var/infinity/snapshot";
+
     // default persistence parameter
-    constexpr std::string_view DEFAULT_PERSISTENCE_DIR = "/var/infinity/persistence"; // Empty means disabled
+    constexpr std::string_view DEFAULT_PERSISTENCE_DIR = "/var/infinity/persistence";
     constexpr std::string_view DEFAULT_PERSISTENCE_OBJECT_SIZE_LIMIT_STR = "128MB";   // 128MB
     constexpr SizeT DEFAULT_PERSISTENCE_OBJECT_SIZE_LIMIT = 128 * 1024lu * 1024lu;    // 128MB
 
@@ -293,6 +297,7 @@ export {
 
     constexpr std::string_view RECORD_RUNNING_QUERY_OPTION_NAME = "record_running_query";
 
+    constexpr std::string_view SNAPSHOT_DIR_OPTION_NAME = "snapshot_dir";
     // Variable name
     constexpr std::string_view QUERY_COUNT_VAR_NAME = "query_count";                         // global and session
     constexpr std::string_view SESSION_COUNT_VAR_NAME = "session_count";                     // global

src/common/utility/exception.cpp

@@ -25,9 +25,23 @@ import logger;
 import third_party;
 import infinity_context;
 import cleanup_scanner;
+import txn_manager;
+import txn_context;
 
 namespace infinity {
 
+void PrintTransactionHistory() {
+    TxnManager *txn_manager = InfinityContext::instance().storage()->txn_manager();
+
+    Vector<SharedPtr<TxnContext>> txn_contexts = txn_manager->GetTxnContextHistories();
+
+    SizeT history_count = txn_contexts.size();
+    for (SizeT idx = 0; idx < history_count; ++idx) {
+        SharedPtr<TxnContext> txn_history = txn_contexts[idx];
+        LOG_CRITICAL(txn_history->ToString());
+    }
+}
+
 void PrintStacktrace(const String &err_msg) {
     int trace_stack_depth = 256;
     void *array[256];
@@ -60,14 +74,20 @@ std::string_view GetErrorMsg(const String &message) {
 }
 
 void UnrecoverableError(const String &message, const char *file_name, u32 line) {
-    // auto *storage = InfinityContext::instance().storage();
+    auto *storage = InfinityContext::instance().storage();
+    if (storage != nullptr) {
+        if (storage->txn_manager() != nullptr) {
+            infinity::PrintTransactionHistory();
+        }
+    }
     // if (storage != nullptr) {
     //     CleanupInfoTracer *cleanup_tracer = storage->cleanup_info_tracer();
     //     String error_msg = cleanup_tracer->GetCleanupInfo();
     //     LOG_ERROR(std::move(error_msg));
     // }
     String location_message = fmt::format("{}@{}:{}", message, infinity::TrimPath(file_name), line);
     if (IS_LOGGER_INITIALIZED()) {
+
         PrintStacktrace(location_message);
     }
     Logger::Flush();

src/common/utility/exception.cppm

@@ -21,6 +21,7 @@ import status;
 namespace infinity {
 
 export void PrintStacktrace(const String &err_msg);
+export void PrintTransactionHistory();
 
 export class RecoverableException : public std::exception {
 public:

src/common/utility/utility.cpp

@@ -15,6 +15,7 @@
 module;
 
 #include <iomanip>
+#include <openssl/md5.h>
 #include <regex>
 #include <sstream>
 
@@ -124,4 +125,32 @@ String StringTransform(const String &source, const String &from, const String &t
     return ret;
 }
 
+String CalcMD5(const char *input_str, SizeT length) {
+    MD5_CTX ctx;
+    MD5_Init(&ctx);
+    MD5_Update(&ctx, input_str, length);
+
+    unsigned char md[MD5_DIGEST_LENGTH];
+    MD5_Final(md, &ctx);
+
+    std::stringstream hex_stream;
+    for (int i = 0; i < MD5_DIGEST_LENGTH; ++i) {
+        hex_stream << std::hex << std::setw(2) << std::setfill('0') << static_cast<int>(md[i]);
+    }
+
+    return hex_stream.str();
+}
+
+String CalcMD5(const String& input_file) {
+    std::ifstream input(input_file, std::ios::binary);
+    input.seekg(0, input.end);
+    size_t raw_file_size = input.tellg();
+    input.seekg(0, input.beg);
+    std::vector<char> raw_data(raw_file_size);
+    input.read(raw_data.data(), raw_file_size);
+    input.close();
+    String md5 = CalcMD5(raw_data.data(), raw_file_size);
+    return md5;
+}
+
 } // namespace infinity

src/common/utility/utility.cppm

@@ -40,4 +40,9 @@ IdentifierValidationStatus IdentifierValidation(const String &identifier);
 bool ParseIPPort(const String &str, String &ip, i64 &port);
 
 String StringTransform(const String &source, const String &from, const String &to);
+
+String CalcMD5(const char* input_str, SizeT length);
+
+String CalcMD5(const String& filename);
+
 } // namespace infinity

src/executor/explain_physical_plan.cpp

@@ -1481,6 +1481,24 @@ void ExplainPhysicalPlan::Explain(const PhysicalShow *show_node, SharedPtr<Vecto
             result->emplace_back(MakeShared<String>(output_columns_str));
             break;
         }
+        case ShowStmtType::kTransactionHistory: {
+            String show_str;
+            if (intent_size != 0) {
+                show_str = String(intent_size - 2, ' ');
+                show_str += "-> SHOW TRANSACTION HISTORY";
+            } else {
+                show_str = "SHOW TRANSACTION HISTORY";
+            }
+            show_str += "(";
+            show_str += std::to_string(show_node->node_id());
+            show_str += ")";
+            result->emplace_back(MakeShared<String>(show_str));
+
+            String output_columns_str = String(intent_size, ' ');
+            output_columns_str += " - output columns: [transactions]";
+            result->emplace_back(MakeShared<String>(output_columns_str));
+            break;
+        }
         case ShowStmtType::kSegments: {
             String show_str;
             if (intent_size != 0) {
@@ -1824,6 +1842,42 @@ void ExplainPhysicalPlan::Explain(const PhysicalShow *show_node, SharedPtr<Vecto
             result->emplace_back(MakeShared<String>(output_columns_str));
             break;
         }
+        case ShowStmtType::kListSnapshots: {
+            String show_str;
+            if (intent_size != 0) {
+                show_str = String(intent_size - 2, ' ');
+                show_str += "-> SHOW SNAPSHOTS ";
+            } else {
+                show_str = "SHOW SNAPSHOTS ";
+            }
+            show_str += "(";
+            show_str += std::to_string(show_node->node_id());
+            show_str += ")";
+            result->emplace_back(MakeShared<String>(show_str));
+
+            String output_columns_str = String(intent_size, ' ');
+            output_columns_str += " - output columns: [name, count, total_size]";
+            result->emplace_back(MakeShared<String>(output_columns_str));
+            break;
+        }
+        case ShowStmtType::kShowSnapshot: {
+            String show_str;
+            if (intent_size != 0) {
+                show_str = String(intent_size - 2, ' ');
+                show_str += "-> SHOW MEMORY ALLOCATION ";
+            } else {
+                show_str = "SHOW MEMORY ALLOCATION ";
+            }
+            show_str += "(";
+            show_str += std::to_string(show_node->node_id());
+            show_str += ")";
+            result->emplace_back(MakeShared<String>(show_str));
+
+            String output_columns_str = String(intent_size, ' ');
+            output_columns_str += " - output columns: [name, count, total_size]";
+            result->emplace_back(MakeShared<String>(output_columns_str));
+            break;
+        }
         case ShowStmtType::kFunction: {
             String show_str;
             if (intent_size != 0) {