Add hash mode for ad-hoc queries.

gz · gz · commit 30d9d73850c3 · 2025-08-22T18:20:26.000Z
Signed-off-by: Gerd Zellweger &lt;mail@gerdzellweger.com&gt;
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -56,6 +56,7 @@ arc-swap = "1.5.1"
 arcstr = "1.2.0"
 arrow = "55"
 arrow-json = "55"
+arrow-digest = "55"
 ascii_table = "=4.0.2"
 async-channel = "2.3.1"
 async-std = "1.12.0"
@@ -90,6 +91,7 @@ crossbeam = "0.8.4"
 crossbeam-utils = "0.8.6"
 csv = "1.2.2"
 csv-core = "0.1.10"
+dashmap = "6.1.0"
 datafusion = "47"
 dbsp = { path = "crates/dbsp", version = "0.129.0" }
 dbsp_nexmark = { path = "crates/nexmark" }
diff --git a/crates/adapters/Cargo.toml b/crates/adapters/Cargo.toml
@@ -183,8 +183,9 @@ backtrace = { workspace = true }
 itertools = { workspace = true }
 metrics-process = { version = "2.4.0", default-features = false }
 serde_json_path_to_error = { workspace = true }
-dashmap = "6.1.0"
 smallstr = { workspace = true }
+dashmap = { workspace = true }
+arrow-digest = { workspace = true }
 
 [package.metadata.cargo-machete]
 ignored = ["num-traits"]
diff --git a/crates/adapters/src/adhoc/executor.rs b/crates/adapters/src/adhoc/executor.rs
@@ -1,6 +1,7 @@
 use arrow::array::RecordBatch;
 use arrow::ipc::writer::StreamWriter;
 use arrow::util::pretty::pretty_format_batches;
+use arrow_digest::{RecordDigest, RecordDigestV0};
 use arrow_json::writer::LineDelimited;
 use arrow_json::WriterBuilder;
 use async_stream::{stream, try_stream};
@@ -18,6 +19,7 @@ use parquet::arrow::async_writer::AsyncFileWriter;
 use parquet::arrow::AsyncArrowWriter;
 use parquet::basic::Compression;
 use parquet::file::properties::WriterProperties;
+use sha2::Sha256;
 use std::convert::Infallible;
 use tokio::sync::mpsc::error::SendError;
 use tokio::sync::oneshot::Receiver;
@@ -96,6 +98,29 @@ pub(crate) fn stream_text_query(
     }
 }
 
+pub(crate) async fn hash_query_result(df: DataFrame) -> Result<String, PipelineError> {
+    let schema = df.schema().inner().clone();
+    let stream_executor = execute_stream(df)
+        .await
+        .map_err(|e| PipelineError::AdHocQueryError {
+            error: e.to_string(),
+            df: None,
+        })?;
+    let mut stream = stream_executor.map_err(|e| PipelineError::AdHocQueryError {
+        error: e.to_string(),
+        df: Some(Box::new(e)),
+    })?;
+
+    let mut digest = RecordDigestV0::<Sha256>::new(&schema);
+    while let Some(batch) = stream.next().await {
+        let batch = batch.map_err(PipelineError::from)?;
+        digest.update(&batch);
+    }
+
+    let hash = digest.finalize();
+    Ok(format!("{:X}", hash))
+}
+
 pub(crate) fn stream_json_query(
     df: DataFrame,
 ) -> impl Stream<Item = Result<ByteString, PipelineError>> {
diff --git a/crates/adapters/src/adhoc/format.rs b/crates/adapters/src/adhoc/format.rs
@@ -2,6 +2,7 @@ use arrow::array::RecordBatch;
 use arrow::error::ArrowError;
 use arrow::util::display::{ArrayFormatter, FormatOptions};
 use comfy_table::{Cell, Table};
+
 pub(crate) fn create_table(results: &[RecordBatch]) -> Result<Table, ArrowError> {
     let options = FormatOptions::default().with_display_error(true);
     let mut table = Table::new();
diff --git a/crates/adapters/src/adhoc/mod.rs b/crates/adapters/src/adhoc/mod.rs
@@ -7,8 +7,8 @@ use datafusion::execution::runtime_env::RuntimeEnvBuilder;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::prelude::*;
 use executor::{
-    infallible_from_bytestring, stream_arrow_query, stream_json_query, stream_parquet_query,
-    stream_text_query,
+    hash_query_result, infallible_from_bytestring, stream_arrow_query, stream_json_query,
+    stream_parquet_query, stream_text_query,
 };
 use feldera_adapterlib::errors::journal::ControllerError;
 use feldera_types::config::PipelineConfig;
@@ -165,6 +165,20 @@ async fn adhoc_query_handler(
                 }
             }
         }
+        AdHocResultFormat::Hash => {
+            let hash_result = hash_query_result(df).await;
+            match hash_result {
+                Ok(hash) => {
+                    ws_session.text(hash).await?;
+                }
+                Err(e) => {
+                    ws_session
+                        .text(serde_json::to_string(&e).unwrap_or(e.to_string()))
+                        .await?;
+                    ws_close(ws_session, CloseCode::Error).await;
+                }
+            }
+        }
     }
 
     Ok(())
@@ -302,5 +316,8 @@ pub(crate) async fn stream_adhoc_result(
                 .content_type(mime::APPLICATION_OCTET_STREAM)
                 .streaming(stream_parquet_query(df)))
         }
+        AdHocResultFormat::Hash => Ok(HttpResponse::Ok()
+            .content_type(mime::TEXT_PLAIN)
+            .body(hash_query_result(df).await?)),
     }
 }
diff --git a/crates/fda/src/adhoc.rs b/crates/fda/src/adhoc.rs
@@ -157,6 +157,7 @@ pub(crate) async fn handle_adhoc_query(
             eprintln!("`query` command does not support Prometheus output format");
             std::process::exit(1);
         }
+        OutputFormat::Hash => AdHocResultFormat::Hash,
     };
     let sql = sql.unwrap_or_else(|| {
         if stdin {
@@ -228,6 +229,19 @@ pub(crate) async fn handle_adhoc_query(
             result_file.flush().unwrap();
             println!("Query result saved to '{}'", path.display());
         }
+        AdHocResultFormat::Hash => {
+            while let Some(chunk) = websocket.next().await {
+                let mut text: String = String::new();
+                if let Ok(Message::Text(chunk)) = chunk {
+                    text.push_str(chunk.as_str());
+                    print!("{}", chunk);
+                } else {
+                    handle_websocket_message_generic(&mut websocket, chunk).await;
+                    break;
+                }
+            }
+            println!()
+        }
     }
 }
 
diff --git a/crates/fda/src/bench/mod.rs b/crates/fda/src/bench/mod.rs
@@ -324,7 +324,10 @@ impl Benchmark {
         match format {
             OutputFormat::Json => serde_json::to_string_pretty(&self.as_map()).unwrap(),
             OutputFormat::Text => self.format_as_text(),
-            OutputFormat::ArrowIpc | OutputFormat::Parquet | OutputFormat::Prometheus => {
+            OutputFormat::ArrowIpc
+            | OutputFormat::Parquet
+            | OutputFormat::Prometheus
+            | OutputFormat::Hash => {
                 warn!("Format '{}' is not supported for benchmark results, falling back to text format", format);
                 self.format_as_text()
             }
diff --git a/crates/fda/src/cli.rs b/crates/fda/src/cli.rs
@@ -126,6 +126,10 @@ pub enum OutputFormat {
     ///
     /// This format can only be specified for the `metrics` command.
     Prometheus,
+    /// Return the output as a hash of the result.
+    ///
+    /// This format can only be specified for SQL queries.
+    Hash,
 }
 
 impl Display for OutputFormat {
@@ -136,6 +140,7 @@ impl Display for OutputFormat {
             OutputFormat::ArrowIpc => "arrow_ipc",
             OutputFormat::Parquet => "parquet",
             OutputFormat::Prometheus => "prometheus",
+            OutputFormat::Hash => "hash",
         };
         write!(f, "{}", output)
     }
diff --git a/crates/fda/src/shell.rs b/crates/fda/src/shell.rs
@@ -153,7 +153,10 @@ pub async fn shell(format: OutputFormat, name: String, client: Client) {
                                 OutputFormat::Json => "json",
                                 OutputFormat::ArrowIpc => "arrow",
                                 OutputFormat::Parquet => "parquet",
-                                OutputFormat::Prometheus => "prometheus",
+                                OutputFormat::Hash => "hash",
+                                OutputFormat::Prometheus => {
+                                    panic!("Prometheus format is not supported for ad-hoc SQL");
+                                }
                             };
                             match client
                                 .pipeline_adhoc_sql()
diff --git a/crates/feldera-types/src/query.rs b/crates/feldera-types/src/query.rs
@@ -22,6 +22,13 @@ pub enum AdHocResultFormat {
     Parquet,
     /// Stream data in the arrow IPC format.
     ArrowIpc,
+    /// Returns a hash of the results instead of the actual data.
+    /// This is useful for verifying the integrity of the data
+    /// without transferring the entire dataset.
+    ///
+    /// Note that for the hash to be meaningful the query must sort data
+    /// in a deterministic way.
+    Hash,
 }
 
 impl Display for AdHocResultFormat {
@@ -31,6 +38,7 @@ impl Display for AdHocResultFormat {
             AdHocResultFormat::Json => write!(f, "json"),
             AdHocResultFormat::Parquet => write!(f, "parquet"),
             AdHocResultFormat::ArrowIpc => write!(f, "arrow_ipc"),
+            AdHocResultFormat::Hash => write!(f, "hash"),
         }
     }
 }
diff --git a/python/README.md b/python/README.md
@@ -5,19 +5,19 @@ Feldera Python is the Feldera SDK for Python developers.
 ## Installation
 
 ```bash
-pip install feldera
+uv pip install feldera
 ```
 
 ### Installing from Github
 
 ```bash
-pip install git+https://github.com/feldera/feldera#subdirectory=python
+uv pip install git+https://github.com/feldera/feldera#subdirectory=python
 ```
 
 Similarly, to install from a specific branch:
 
 ```bash
-$ pip install git+https://github.com/feldera/feldera@{BRANCH_NAME}#subdirectory=python
+uv pip install git+https://github.com/feldera/feldera@{BRANCH_NAME}#subdirectory=python
 ```
 
 Replace `{BRANCH_NAME}` with the name of the branch you want to install from.
@@ -28,7 +28,12 @@ If you have cloned the Feldera repo, you can install the python SDK as follows:
 
 ```bash
 # the Feldera Python SDK is present inside the python/ directory
-pip install python/
+cd python
+# If you don't have a virtual environment, create one
+uv venv
+source .venv/activate
+# Install the SDK in editable mode
+uv pip install .
 ```
 
 ## Documentation
@@ -38,7 +43,7 @@ The Python SDK documentation is available at
 
 To build the html documentation run:
 
-Ensure that you have sphinx installed. If not, install it using `pip install sphinx`.
+Ensure that you have sphinx installed. If not, install it using `uv pip install sphinx`.
 
 Then run the following commands:
 
@@ -69,7 +74,13 @@ To run tests from a specific file:
 (cd python && python3 -m pytest ./tests/path-to-file.py)
 ```
 
-#### Running Tests
+To run a specific test:
+
+```bash
+uv run python -m pytest tests/test_shared_pipeline.py::TestPipeline::test_adhoc_query_hash -v
+```
+
+#### Running All Tests
 
 The tests validate end-to-end correctness of SQL functionality.  To
 run the tests use:
diff --git a/python/feldera/pipeline.py b/python/feldera/pipeline.py
@@ -809,6 +809,25 @@ def query_tabular(self, query: str) -> Generator[str, None, None]:
 
         return self.client.query_as_text(self.name, query)
 
+    def query_hash(self, query: str):
+        """
+        Executes an ad-hoc SQL query on this pipeline and returns the result
+        as a hash of the result set. This is useful for quickly checking
+        if the result set has changed without retrieving the entire result.
+
+        Note:
+            For a stable hash, the query must be deterministic which means
+            it should be sorted.
+
+        :param query: The SQL query to be executed.
+
+        :raises FelderaAPIError: If the pipeline is not in a RUNNING or PAUSED
+            state.
+        :raises FelderaAPIError: If querying a non materialized table or view.
+        :raises FelderaAPIError: If the query is invalid.
+        """
+        return self.client.query_as_hash(self.name, query)
+
     def execute(self, query: str):
         """
         Executes an ad-hoc SQL query on the current pipeline, discarding its
diff --git a/python/feldera/rest/feldera_client.py b/python/feldera/rest/feldera_client.py
@@ -895,6 +895,30 @@ def query_as_text(
             if chunk:
                 yield chunk.decode("utf-8")
 
+    def query_as_hash(
+        self, pipeline_name: str, query: str
+    ) -> str:
+        """
+        Executes an ad-hoc query on the specified pipeline and returns a hash of the result.
+
+        :param pipeline_name: The name of the pipeline to query.
+        :param query: The SQL query to be executed.
+        :return: A generator yielding the query result in tabular format, one line at a time.
+        """
+        params = {
+            "pipeline_name": pipeline_name,
+            "sql": query,
+            "format": "hash",
+        }
+
+        resp = self.http.get(
+            path=f"/pipelines/{pipeline_name}/query",
+            params=params,
+            stream=False,
+        )
+        return resp
+
+
     def query_as_parquet(self, pipeline_name: str, query: str, path: str):
         """
         Executes an ad-hoc query on the specified pipeline and saves the result to a parquet file.
diff --git a/python/tests/test_shared_pipeline.py b/python/tests/test_shared_pipeline.py

Original file line number	Diff line number	Diff line change
`@@ -7,8 +7,8 @@ use datafusion::execution::runtime_env::RuntimeEnvBuilder;`
`7`	`7`	`use datafusion::execution::SessionStateBuilder;`
`8`	`8`	`use datafusion::prelude::*;`
`9`	`9`	`use executor::{`
`10`		`- infallible_from_bytestring, stream_arrow_query, stream_json_query, stream_parquet_query,`
`11`		`- stream_text_query,`
	`10`	`+ hash_query_result, infallible_from_bytestring, stream_arrow_query, stream_json_query,`
	`11`	`+ stream_parquet_query, stream_text_query,`
`12`	`12`	`};`
`13`	`13`	`use feldera_adapterlib::errors::journal::ControllerError;`
`14`	`14`	`use feldera_types::config::PipelineConfig;`
`@@ -165,6 +165,20 @@ async fn adhoc_query_handler(`
`165`	`165`	`}`
`166`	`166`	`}`
`167`	`167`	`}`
	`168`	`+ AdHocResultFormat::Hash => {`
	`169`	`+ let hash_result = hash_query_result(df).await;`
	`170`	`+ match hash_result {`
	`171`	`+ Ok(hash) => {`
	`172`	`+ ws_session.text(hash).await?;`
	`173`	`+ }`
	`174`	`+ Err(e) => {`
	`175`	`+ ws_session`
	`176`	`+ .text(serde_json::to_string(&e).unwrap_or(e.to_string()))`
	`177`	`+ .await?;`
	`178`	`+ ws_close(ws_session, CloseCode::Error).await;`
	`179`	`+ }`
	`180`	`+ }`
	`181`	`+ }`
`168`	`182`	`}`
`169`	`183`
`170`	`184`	`Ok(())`
`@@ -302,5 +316,8 @@ pub(crate) async fn stream_adhoc_result(`
`302`	`316`	`.content_type(mime::APPLICATION_OCTET_STREAM)`
`303`	`317`	`.streaming(stream_parquet_query(df)))`
`304`	`318`	`}`
	`319`	`+ AdHocResultFormat::Hash => Ok(HttpResponse::Ok()`
	`320`	`+ .content_type(mime::TEXT_PLAIN)`
	`321`	`+ .body(hash_query_result(df).await?)),`
`305`	`322`	`}`
`306`	`323`	`}`