chore: add new API to chdb.h

Author: wudidapaopao

programs/local/LocalChdb.h

@@ -125,6 +125,22 @@ class local_result_wrapper
         }
         return result->bytes_read;
     }
+    size_t storage_rows_read()
+    {
+        if (result == nullptr)
+        {
+            return 0;
+        }
+        return result->storage_rows_read;
+    }
+    size_t storage_bytes_read()
+    {
+        if (result == nullptr)
+        {
+            return 0;
+        }
+        return result->storage_bytes_read;
+    }
     double elapsed()
     {
         if (result == nullptr)
@@ -166,6 +182,8 @@ class query_result
     size_t size() { return result_wrapper->size(); }
     size_t rows_read() { return result_wrapper->rows_read(); }
     size_t bytes_read() { return result_wrapper->bytes_read(); }
+    size_t storgae_rows_read() { return result_wrapper->storage_rows_read(); }
+    size_t storage_bytes_read() { return result_wrapper->storage_bytes_read(); }
     double elapsed() { return result_wrapper->elapsed(); }
     bool has_error() { return result_wrapper->has_error(); }
     py::str error_message() { return result_wrapper->error_message(); }

programs/local/LocalServer.cpp

@@ -1289,6 +1289,8 @@ static local_result_v2 * createMaterializedLocalQueryResult(DB::LocalServer * se
             }
             result->rows_read = server->getProcessedRows();
             result->bytes_read = server->getProcessedBytes();
+            result->storage_rows_read = server->getStorgaeRowsRead();
+            result->storage_bytes_read = server->getStorageBytesRead();
             result->elapsed = server->getElapsedTime();
         }
     }

programs/local/chdb.h

@@ -1,16 +1,19 @@
 #pragma once
 
 #ifdef __cplusplus
-#    include <cstddef>
-#    include <cstdint>
+#include <cstddef>
+#include <cstdint>
 extern "C" {
 #else
-#    include <stdbool.h>
-#    include <stddef.h>
-#    include <stdint.h>
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
 #endif
 
 #define CHDB_EXPORT __attribute__((visibility("default")))
+
+#ifndef CHDB_NO_DEPRECATED
+// WARNING: The following structs are deprecated and will be removed in a future version.
 struct local_result
 {
     char * buf;
@@ -45,12 +48,6 @@ struct local_result_v2
 };
 #endif
 
-CHDB_EXPORT struct local_result * query_stable(int argc, char ** argv);
-CHDB_EXPORT void free_result(struct local_result * result);
-
-CHDB_EXPORT struct local_result_v2 * query_stable_v2(int argc, char ** argv);
-CHDB_EXPORT void free_result_v2(struct local_result_v2 * result);
-
 /**
  * Connection structure for chDB
  * Contains server instance, connection state, and query processing queue
@@ -62,10 +59,37 @@ struct chdb_conn
     void * queue; /* Query processing queue */
 };
 
+#endif
+
+// Opaque handle for query results.
+// Internal data structure managed by chDB implementation.
+// Users should only interact through API functions.
+typedef struct {
+	void * internal_data;
+} chdb_result;
+
+// Connection handle wrapping database session state.
+// Internal data structure managed by chDB implementation.
+// Users should only interact through API functions.
+typedef struct _chdb_connection {
+	void * internal_data;
+} * chdb_connection;
+
+// Opaque handle for streaming query results.
+// Internal data structure managed by chDB implementation.
+// Users should only interact through API functions.
 typedef struct {
 	void * internal_data;
 } chdb_streaming_result;
 
+#ifndef CHDB_NO_DEPRECATED
+// WARNING: The following interfaces are deprecated and will be removed in a future version.
+CHDB_EXPORT struct local_result * query_stable(int argc, char ** argv);
+CHDB_EXPORT void free_result(struct local_result * result);
+
+CHDB_EXPORT struct local_result_v2 * query_stable_v2(int argc, char ** argv);
+CHDB_EXPORT void free_result_v2(struct local_result_v2 * result);
+
 /**
  * Creates a new chDB connection.
  * Only one active connection is allowed per process.
@@ -110,12 +134,67 @@ CHDB_EXPORT struct local_result_v2 * query_conn(struct chdb_conn * conn, const c
 CHDB_EXPORT chdb_streaming_result * query_conn_streaming(struct chdb_conn * conn, const char * query, const char * format);
 
 /**
- * Retrieves error message from streaming result.
- * @brief Gets error message associated with streaming query execution
+ * Fetches next chunk of streaming results.
+ * @brief Iterates through streaming query results
+ * @param conn Active connection handle
  * @param result Streaming result handle from query_conn_streaming()
- * @return Null-terminated error message string, or NULL if no error occurred
+ * @return Materialized result chunk with data
+ * @note Returns empty result when stream ends
  */
-CHDB_EXPORT const char * chdb_streaming_result_error(chdb_streaming_result * result);
+CHDB_EXPORT struct local_result_v2 * chdb_streaming_fetch_result(struct chdb_conn * conn, chdb_streaming_result * result);
+
+/**
+ * Cancels ongoing streaming query.
+ * @brief Aborts streaming query execution and cleans up resources
+ * @param conn Active connection handle
+ * @param result Streaming result handle to cancel
+ */
+CHDB_EXPORT void chdb_streaming_cancel_query(struct chdb_conn * conn, chdb_streaming_result * result);
+
+#endif
+
+/**
+ * Creates a new chDB connection.
+ * Only one active connection is allowed per process.
+ * Creating a new connection with different path requires closing existing connection.
+ *
+ * @param argc Number of command-line arguments
+ * @param argv Command-line arguments array (--path=<db_path> to specify database location)
+ * @return Pointer to connection pointer, or NULL on failure
+ * @note Default path is ":memory:" if not specified
+ */
+CHDB_EXPORT chdb_connection * chdb_connect(int argc, char ** argv);
+
+/**
+ * Closes an existing chDB connection and cleans up resources.
+ * Thread-safe function that handles connection shutdown and cleanup.
+ *
+ * @param conn Pointer to connection pointer to close
+ */
+ CHDB_EXPORT void chdb_close_conn(chdb_connection * conn);
+
+/**
+ * Executes a query on the given connection.
+ * Thread-safe function that handles query execution in a separate thread.
+ *
+ * @param conn Connection to execute query on
+ * @param query SQL query string to execute
+ * @param format Output format string (e.g., "CSV", default format)
+ * @return Query result structure containing output or error message
+ * @note Returns error result if connection is invalid or closed
+ */
+CHDB_EXPORT chdb_result * chdb_query(chdb_connection conn, const char * query, const char * format);
+
+/**
+ * Executes a streaming query on the given connection.
+ * @brief Initializes streaming query execution and returns result handle
+ * @param conn Connection to execute query on
+ * @param query SQL query string to execute
+ * @param format Output format string (e.g. "CSV", default format)
+ * @return Streaming result handle containing query state or error message
+ * @note Returns error result if connection is invalid or closed
+ */
+CHDB_EXPORT chdb_streaming_result * chdb_stream_query(chdb_connection conn, const char * query, const char * format);
 
 /**
  * Fetches next chunk of streaming results.
@@ -125,15 +204,23 @@ CHDB_EXPORT const char * chdb_streaming_result_error(chdb_streaming_result * res
  * @return Materialized result chunk with data
  * @note Returns empty result when stream ends
  */
-CHDB_EXPORT struct local_result_v2 * chdb_streaming_fetch_result(struct chdb_conn * conn, chdb_streaming_result * result);
+CHDB_EXPORT chdb_result * chdb_stream_fetch_result(chdb_connection conn, chdb_streaming_result * result);
 
 /**
  * Cancels ongoing streaming query.
  * @brief Aborts streaming query execution and cleans up resources
  * @param conn Active connection handle
  * @param result Streaming result handle to cancel
  */
-CHDB_EXPORT void chdb_streaming_cancel_query(struct chdb_conn * conn, chdb_streaming_result * result);
+CHDB_EXPORT void chdb_stream_cancel_query(chdb_connection conn, chdb_streaming_result * result);
+
+/**
+ * Retrieves error message from streaming result.
+ * @brief Gets error message associated with streaming query execution
+ * @param result Streaming result handle from query_conn_streaming()
+ * @return Null-terminated error message string, or NULL if no error occurred
+ */
+CHDB_EXPORT const char * chdb_streaming_result_error(chdb_streaming_result * result);
 
 /**
  * Releases resources associated with streaming result.

src/Client/ClientBase.h

@@ -127,6 +127,9 @@ class ClientBase
         return result;
     }
 
+    size_t getStorgaeRowsRead() const { return connection->getCHDBProgress().read_rows; }
+    size_t getStorageBytesRead() const { return connection->getCHDBProgress().read_bytes; }
+
     size_t getProcessedRows() const { return processed_rows; }
     size_t getProcessedBytes() const { return processed_bytes; }
     double getElapsedTime() const { return progress_indication.elapsedSeconds(); }

src/Client/LocalConnection.cpp

@@ -43,6 +43,8 @@ LocalConnection::LocalConnection(ContextPtr context_, ReadBuffer * in_, bool sen
     /// Authenticate and create a context to execute queries.
     session.authenticate("default", "", Poco::Net::SocketAddress{});
     session.makeSessionContext();
+
+    send_progress = false;
 }
 
 LocalConnection::~LocalConnection()
@@ -77,6 +79,11 @@ void LocalConnection::updateProgress(const Progress & value)
     state->progress.incrementPiecewiseAtomically(value);
 }
 
+void LocalConnection::updateCHDBProgress(const Progress & value)
+{
+    chdb_progress.incrementPiecewiseAtomically(value);
+}
+
 void LocalConnection::sendProfileEvents()
 {
     Block profile_block;
@@ -113,6 +120,11 @@ void LocalConnection::sendQuery(
         query_context->setProgressCallback([this] (const Progress & value) { this->updateProgress(value); });
         query_context->setFileProgressCallback([this](const FileProgress & value) { this->updateProgress(Progress(value)); });
     }
+    else
+    {
+        query_context->setProgressCallback([this] (const Progress & value) { this->updateCHDBProgress(value); });
+        query_context->setFileProgressCallback([this](const FileProgress & value) { this->updateCHDBProgress(Progress(value)); });
+    }
 
     /// Switch the database to the desired one (set by the USE query)
     /// but don't attempt to do it if we are already in that database.
@@ -125,6 +137,7 @@ void LocalConnection::sendQuery(
 
     state.reset();
     state.emplace();
+    chdb_progress.reset();
 
     state->query_id = query_id;
     state->query = query;

src/Client/LocalConnection.h

@@ -142,6 +142,7 @@ class LocalConnection : public IServerConnection, WithContext
 
     void setThrottler(const ThrottlerPtr &) override {}
 
+    const Progress & getCHDBProgress() const { return chdb_progress; }
 #if USE_PYTHON
     void resetQueryContext();
 #endif
@@ -153,6 +154,8 @@ class LocalConnection : public IServerConnection, WithContext
 
     void updateProgress(const Progress & value);
 
+    void updateCHDBProgress(const Progress & value);
+
     void sendProfileEvents();
 
     /// Returns true on executor timeout, meaning a retryable error.
@@ -170,6 +173,8 @@ class LocalConnection : public IServerConnection, WithContext
 
     std::optional<LocalQueryState> state;
 
+    Progress chdb_progress;
+
     /// Last "server" packet.
     std::optional<UInt64> next_packet_type;