feat(dynamodb): support querying, editing, and exporting

crates/dbx-core/Cargo.toml

@@ -5,8 +5,9 @@ edition = "2021"
 license = "Apache-2.0"
 
 [features]
-default = ["duckdb-sidecar", "mq-admin", "sqlite-sqlcipher", "system-fonts"]
+default = ["duckdb-sidecar", "dynamodb", "mq-admin", "sqlite-sqlcipher", "system-fonts"]
 duckdb-sidecar = []
+dynamodb = ["dep:aws-sdk-dynamodb"]
 mq-admin = ["dep:rumqttc"]
 sqlite-sqlcipher = ["rusqlite/bundled-sqlcipher-vendored-openssl"]
 openapi = ["dep:utoipa"]
@@ -50,6 +51,7 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "str
 prometheus-parse = "=0.2.5"
 futures = "0.3"
 mongodb = { version = "3.2.5", features = ["socks5-proxy"] }
+aws-sdk-dynamodb = { version = "=1.104.0", optional = true }
 rumqttc = { version = "0.24", features = ["websocket"], optional = true }
 russh = "0.60"
 pageant = "0.2"

crates/dbx-core/assets/database-drivers.manifest.json

@@ -296,6 +296,35 @@
         "driverManagement": false
       }
     },
+    {
+      "dbType": "dynamodb",
+      "label": "Amazon DynamoDB",
+      "runtimeMode": "native",
+      "mcpMode": "bridge",
+      "singleConnectionPool": true,
+      "metadataConnectionScoped": false,
+      "skipTcpProbe": false,
+      "defaultPort": 8000,
+      "supportLevel": "operate",
+      "capabilities": {
+        "queryExecution": false,
+        "metadataBrowse": true,
+        "objectBrowser": false,
+        "objectSource": false,
+        "schemaSearch": false,
+        "diagram": false,
+        "tableDataEdit": true,
+        "tableStructureEdit": false,
+        "tableImport": false,
+        "dataTransfer": false,
+        "sqlFileExecution": false,
+        "databaseCreate": false,
+        "fieldLineage": false,
+        "sqlExplain": false,
+        "userAdmin": false,
+        "driverManagement": false
+      }
+    },
     {
       "dbType": "oracle",
       "label": "Oracle",

crates/dbx-core/src/connection.rs

@@ -87,6 +87,7 @@ pub enum PoolKind {
     Redis(db::redis_driver::RedisConnection),
     DuckDbWorker(DuckDbWorkerHandle),
     MongoDb(mongodb::Client),
+    DynamoDb(db::dynamodb_driver::DynamoDbClient),
     ClickHouse(db::clickhouse_driver::ChClient),
     SqlServer(Arc<tokio::sync::Mutex<db::sqlserver::SqlServerClient>>),
     Elasticsearch(db::elasticsearch_driver::EsClient),
@@ -2022,6 +2023,11 @@ impl AppState {
                     }
                 }
             }
+            DatabaseType::DynamoDb => {
+                let client = db::dynamodb_driver::connect(&db_config, &host, port)?;
+                db::dynamodb_driver::test_connection(&client, connect_timeout).await?;
+                PoolKind::DynamoDb(client)
+            }
             DatabaseType::ClickHouse => {
                 let username = if db_config.username.is_empty() { None } else { Some(db_config.username.clone()) };
                 let password = if db_config.password.is_empty() { None } else { Some(db_config.password.clone()) };
@@ -3100,6 +3106,18 @@ impl AppState {
                         }
                     }
                 }
+                PoolKind::DynamoDb(client) => {
+                    let client = client.clone();
+                    drop(connections);
+                    let timeout = crate::db::connection_timeout();
+                    match db::dynamodb_driver::test_connection(&client, timeout).await {
+                        Ok(()) => false,
+                        Err(err) => {
+                            log::warn!("DynamoDB connection pool '{pool_key}' is stale: {err}");
+                            true
+                        }
+                    }
+                }
                 PoolKind::ClickHouse(client) => {
                     let client = client.clone();
                     drop(connections);
@@ -4051,6 +4069,13 @@ impl AppState {
                         false
                     }
                 },
+                PoolKind::DynamoDb(client) => match db::dynamodb_driver::test_connection(client, timeout).await {
+                    Ok(()) => true,
+                    Err(e) => {
+                        log::warn!("DynamoDB connection pool '{key}' is unhealthy: {e}");
+                        false
+                    }
+                },
                 PoolKind::ClickHouse(client) => match db::clickhouse_driver::test_connection(client, timeout).await {
                     Ok(()) => true,
                     Err(e) => {
@@ -4844,6 +4869,7 @@ fn clone_pool_kind(pool: &PoolKind) -> PoolKind {
         #[cfg(not(feature = "duckdb-sidecar"))]
         PoolKind::DuckDbWorker(_) => PoolKind::DuckDbWorker(()),
         PoolKind::MongoDb(client) => PoolKind::MongoDb(client.clone()),
+        PoolKind::DynamoDb(client) => PoolKind::DynamoDb(client.clone()),
         PoolKind::ClickHouse(client) => PoolKind::ClickHouse(client.clone()),
         PoolKind::SqlServer(client) => PoolKind::SqlServer(client.clone()),
         PoolKind::Elasticsearch(client) => PoolKind::Elasticsearch(client.clone()),
@@ -4888,6 +4914,9 @@ async fn close_pool_kind(pool: PoolKind) -> Result<(), String> {
         PoolKind::MongoDb(client) => {
             drop(client);
         }
+        PoolKind::DynamoDb(client) => {
+            drop(client);
+        }
         PoolKind::ClickHouse(client) => {
             drop(client);
         }

crates/dbx-core/src/database_capabilities.rs

@@ -18,6 +18,7 @@ pub fn is_single_connection_pool(db_type: &DatabaseType) -> bool {
             | DatabaseType::Turso
             | DatabaseType::CloudflareD1
             | DatabaseType::MongoDb
+            | DatabaseType::DynamoDb
             | DatabaseType::Hbase
             | DatabaseType::Oracle
             | DatabaseType::Dameng

crates/dbx-core/src/db/document_result.rs

@@ -13,6 +13,10 @@ pub struct DocumentQueryResult {
     // unchanged and only send the flag when Elasticsearch reports a lower bound.
     #[serde(default = "default_total_is_exact", skip_serializing_if = "is_true")]
     pub total_is_exact: bool,
+    /// Opaque document-store cursor for APIs such as DynamoDB that do not
+    /// support offset pagination.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub next_cursor: Option<String>,
 }
 
 fn default_total_is_exact() -> bool {
@@ -35,6 +39,7 @@ mod tests {
             extended_documents: None,
             total: 1,
             total_is_exact: true,
+            next_cursor: None,
         };
 
         let serialized = serde_json::to_value(result).unwrap();

crates/dbx-core/src/db/dynamodb_driver_disabled.rs

@@ -0,0 +1,103 @@
+use std::time::Duration;
+
+use crate::db::document_result::DocumentQueryResult;
+use crate::models::connection::ConnectionConfig;
+use crate::types::QueryResult;
+
+const FEATURE_DISABLED: &str = "DynamoDB support is not included in this DBX build";
+
+#[derive(Clone)]
+pub struct DynamoDbClient {
+    pub region: String,
+    pub endpoint_url: String,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct DynamoDbKeyInfo {
+    pub name: String,
+    pub attribute_type: String,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct DynamoDbIndexInfo {
+    pub name: String,
+    pub kind: String,
+    pub partition_key: DynamoDbKeyInfo,
+    pub sort_key: Option<DynamoDbKeyInfo>,
+    pub projection_type: String,
+    pub non_key_attributes: Vec<String>,
+}
+
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct DynamoDbTableDescription {
+    pub name: String,
+    pub status: String,
+    pub item_count: u64,
+    pub size_bytes: u64,
+    pub partition_key: DynamoDbKeyInfo,
+    pub sort_key: Option<DynamoDbKeyInfo>,
+    pub indexes: Vec<DynamoDbIndexInfo>,
+}
+
+pub fn connect(_config: &ConnectionConfig, _host: &str, _port: u16) -> Result<DynamoDbClient, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn test_connection(_client: &DynamoDbClient, _timeout: Duration) -> Result<(), String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn execute_statement(
+    _client: &DynamoDbClient,
+    _source: &str,
+    _max_rows: usize,
+) -> Result<QueryResult, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn list_tables(_client: &DynamoDbClient) -> Result<Vec<String>, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn describe_table(_client: &DynamoDbClient, _table_name: &str) -> Result<DynamoDbTableDescription, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn find_items(
+    _client: &DynamoDbClient,
+    _table_name: &str,
+    _limit: i64,
+    _filter_json: Option<&str>,
+    _sort_json: Option<&str>,
+    _cursor: Option<&str>,
+) -> Result<DocumentQueryResult, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn count_items(
+    _client: &DynamoDbClient,
+    _table_name: &str,
+    _filter_json: Option<&str>,
+) -> Result<u64, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn insert_item(_client: &DynamoDbClient, _table_name: &str, _doc_json: &str) -> Result<String, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn update_item(
+    _client: &DynamoDbClient,
+    _table_name: &str,
+    _id: &str,
+    _doc_json: &str,
+) -> Result<u64, String> {
+    Err(FEATURE_DISABLED.to_string())
+}
+
+pub async fn delete_item(_client: &DynamoDbClient, _table_name: &str, _id: &str) -> Result<u64, String> {
+    Err(FEATURE_DISABLED.to_string())
+}

crates/dbx-core/src/db/elasticsearch_driver.rs

@@ -829,6 +829,7 @@ fn search_response_to_document_result(result: SearchResponse) -> Result<Document
         extended_documents: None,
         total,
         total_is_exact,
+        next_cursor: None,
     })
 }
 

crates/dbx-core/src/db/meilisearch_driver.rs

@@ -274,6 +274,7 @@ pub async fn find_documents(
         extended_documents: None,
         total: result.total,
         total_is_exact: true,
+        next_cursor: None,
     })
 }
 

crates/dbx-core/src/db/mod.rs

@@ -11,6 +11,12 @@ pub mod duckdb_sql;
 pub mod duckdb_worker_process;
 #[cfg(feature = "duckdb-sidecar")]
 pub mod duckdb_worker_protocol;
+#[cfg(feature = "dynamodb")]
+#[path = "dynamodb_driver.rs"]
+pub mod dynamodb_driver;
+#[cfg(not(feature = "dynamodb"))]
+#[path = "dynamodb_driver_disabled.rs"]
+pub mod dynamodb_driver;
 pub mod easysearch_driver;
 pub mod elasticsearch_driver;
 pub mod elasticsearch_sql;

crates/dbx-core/src/db/mongo_driver.rs

@@ -189,6 +189,7 @@ pub async fn run_command(client: &Client, database: &str, command_json: &str) ->
         extended_documents: Some(vec![extended_document]),
         total: 1,
         total_is_exact: true,
+        next_cursor: None,
     })
 }
 
@@ -1198,6 +1199,7 @@ async fn find_documents_with_total(
         extended_documents: Some(extended_documents),
         total,
         total_is_exact,
+        next_cursor: None,
     })
 }
 
@@ -1354,6 +1356,7 @@ pub async fn find_documents_extended_json(
         raw_documents: None,
         total,
         total_is_exact,
+        next_cursor: None,
     })
 }
 
@@ -1422,6 +1425,7 @@ pub async fn aggregate_documents(
             extended_documents: Some(vec![extended]),
             total: 1,
             total_is_exact: true,
+            next_cursor: None,
         });
     }
 
@@ -1479,6 +1483,7 @@ async fn drain_document_cursor(
         extended_documents: Some(extended_documents),
         total,
         total_is_exact: true,
+        next_cursor: None,
     })
 }
 
@@ -1539,6 +1544,7 @@ pub async fn distinct(
         extended_documents: Some(extended_documents),
         total,
         total_is_exact: true,
+        next_cursor: None,
     })
 }
 
@@ -2107,6 +2113,7 @@ fn single_document_result(document: Option<Document>) -> MongoDocumentResult {
             extended_documents: Some(vec![Bson::Document(document).into_canonical_extjson()]),
             total: 1,
             total_is_exact: true,
+            next_cursor: None,
         },
         None => MongoDocumentResult {
             documents: Vec::new(),
@@ -2114,6 +2121,7 @@ fn single_document_result(document: Option<Document>) -> MongoDocumentResult {
             extended_documents: Some(Vec::new()),
             total: 0,
             total_is_exact: true,
+            next_cursor: None,
         },
     }
 }

crates/dbx-core/src/db/vector_driver.rs

@@ -621,6 +621,7 @@ pub async fn find_documents(
             extended_documents: None,
             total: result.affected_rows,
             total_is_exact: true,
+            next_cursor: None,
         });
     }
 
@@ -669,6 +670,7 @@ pub async fn find_documents(
         extended_documents: None,
         total: result.affected_rows,
         total_is_exact: true,
+        next_cursor: None,
     })
 }
 

crates/dbx-core/src/document_ops.rs

@@ -1,7 +1,7 @@
 use crate::connection::{AppState, PoolKind};
 use crate::db::agent_driver::mongo_document_id_params;
 use crate::db::document_result::DocumentQueryResult;
-use crate::db::{easysearch_driver, elasticsearch_driver, mongo_driver, vector_driver};
+use crate::db::{dynamodb_driver, easysearch_driver, elasticsearch_driver, mongo_driver, vector_driver};
 
 pub use crate::db::vector_driver::CollectionInfo;
 
@@ -54,6 +54,7 @@ pub async fn list_databases_core(state: &AppState, connection_id: &str) -> Resul
             }
             Err(error) => Err(error),
         },
+        PoolKind::DynamoDb(client) => Ok(vec![client.region.clone()]),
         PoolKind::Elasticsearch(_) => Ok(vec!["default".to_string()]),
         PoolKind::Easysearch(_) => Ok(vec!["default".to_string()]),
         PoolKind::Meilisearch(_) => Ok(vec!["default".to_string()]),
@@ -246,6 +247,20 @@ pub async fn list_collections_core(
             infos.extend(specs.into_iter().map(|spec| mongo_collection_info(spec.name, spec.kind)));
             Ok(infos)
         }
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            drop(connections);
+            let names = dynamodb_driver::list_tables(&client).await?;
+            Ok(names
+                .into_iter()
+                .map(|name| CollectionInfo {
+                    id: name.clone(),
+                    name,
+                    kind: Some("table".to_string()),
+                    ..Default::default()
+                })
+                .collect())
+        }
         PoolKind::Elasticsearch(client) => {
             let names = sort_names(elasticsearch_driver::list_indices(client).await?);
             Ok(names.into_iter().map(|n| CollectionInfo { name: n.clone(), id: n, ..Default::default() }).collect())
@@ -408,6 +423,7 @@ pub async fn find_documents_core(
     projection: Option<&str>,
     sort: Option<&str>,
     collation: Option<&str>,
+    cursor: Option<&str>,
 ) -> Result<DocumentQueryResult, String> {
     ensure_document_pool(state, connection_id).await?;
     let connections = state.connections.read().await;
@@ -420,6 +436,12 @@ pub async fn find_documents_core(
             )
             .await
         }
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            drop(connections);
+            let _ = (database, skip, projection, collation);
+            dynamodb_driver::find_items(&client, collection, limit, filter, sort, cursor).await
+        }
         PoolKind::Elasticsearch(client) => {
             let client = client.clone();
             drop(connections);
@@ -469,6 +491,51 @@ pub async fn find_documents_core(
     }
 }
 
+pub async fn count_document_store_documents_core(
+    state: &AppState,
+    connection_id: &str,
+    collection: &str,
+    filter: Option<&str>,
+) -> Result<u64, String> {
+    ensure_document_pool(state, connection_id).await?;
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            drop(connections);
+            dynamodb_driver::count_items(&client, collection, filter).await
+        }
+        PoolKind::Elasticsearch(client) => {
+            let client = client.clone();
+            drop(connections);
+            elasticsearch_driver::count_documents(&client, collection, filter).await
+        }
+        PoolKind::Easysearch(client) => {
+            let client = client.clone();
+            drop(connections);
+            easysearch_driver::count_documents(&client, collection, filter).await
+        }
+        _ => Err("Document count is not supported for this connection".to_string()),
+    }
+}
+
+pub async fn describe_dynamodb_table_core(
+    state: &AppState,
+    connection_id: &str,
+    table: &str,
+) -> Result<dynamodb_driver::DynamoDbTableDescription, String> {
+    ensure_document_pool(state, connection_id).await?;
+    let connections = state.connections.read().await;
+    match connections.get(connection_id).ok_or("Not found")? {
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            drop(connections);
+            dynamodb_driver::describe_table(&client, table).await
+        }
+        _ => Err("Not a DynamoDB connection".to_string()),
+    }
+}
+
 pub async fn count_elasticsearch_documents_core(
     state: &AppState,
     connection_id: &str,
@@ -509,6 +576,11 @@ pub async fn insert_document_core(
     let connections = state.connections.read().await;
     match connections.get(connection_id).ok_or("Not found")? {
         PoolKind::MongoDb(client) => mongo_driver::insert_document(client, database, collection, doc_json).await,
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            drop(connections);
+            dynamodb_driver::insert_item(&client, collection, doc_json).await
+        }
         PoolKind::Elasticsearch(client) => {
             let client = client.clone();
             drop(connections);
@@ -573,6 +645,11 @@ pub async fn update_document_core(
     let connections = state.connections.read().await;
     match connections.get(connection_id).ok_or("Not found")? {
         PoolKind::MongoDb(client) => mongo_driver::update_document(client, database, collection, id, doc_json).await,
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            drop(connections);
+            dynamodb_driver::update_item(&client, collection, id, doc_json).await
+        }
         PoolKind::Elasticsearch(client) => {
             let client = client.clone();
             drop(connections);
@@ -630,6 +707,11 @@ pub async fn delete_document_core_with_type(
     let connections = state.connections.read().await;
     match connections.get(connection_id).ok_or("Not found")? {
         PoolKind::MongoDb(client) => mongo_driver::delete_document(client, database, collection, id).await,
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            drop(connections);
+            dynamodb_driver::delete_item(&client, collection, id).await
+        }
         PoolKind::Elasticsearch(client) => {
             let client = client.clone();
             drop(connections);

crates/dbx-core/src/lib.rs

@@ -1,3 +1,5 @@
+#![recursion_limit = "256"]
+
 pub mod agent_catalog;
 pub mod agent_connection;
 pub mod agent_events;

crates/dbx-core/src/models/connection.rs

@@ -511,6 +511,8 @@ pub enum DatabaseType {
     SqlServer,
     #[serde(rename = "mongodb")]
     MongoDb,
+    #[serde(rename = "dynamodb")]
+    DynamoDb,
     #[serde(rename = "oracle")]
     Oracle,
     #[serde(rename = "elasticsearch")]
@@ -1093,6 +1095,10 @@ impl ConnectionConfig {
                 let db_part = mongo_uri_db_part_for_suffix(&db_part, &suffix);
                 format!("mongodb://{host}:{port}{db_part}{suffix}")
             }
+            DatabaseType::DynamoDb => {
+                let scheme = if self.ssl { "https" } else { "http" };
+                format!("{scheme}://{host}:{port}")
+            }
             DatabaseType::Oracle => format!("oracle://{host}:{port}{db_part}"),
             DatabaseType::Elasticsearch
             | DatabaseType::Easysearch
@@ -1258,6 +1264,10 @@ impl ConnectionConfig {
                     format!("mongodb://{username}:{password}@{host}:{port}{db_part}{suffix}")
                 }
             }
+            DatabaseType::DynamoDb => {
+                let scheme = if self.ssl { "https" } else { "http" };
+                format!("{scheme}://{host}:{port}")
+            }
             DatabaseType::Oracle => {
                 format!("oracle://{}:{}@{host}:{port}{db_part}", username, password)
             }

crates/dbx-core/src/mongo_ops.rs

@@ -224,6 +224,7 @@ pub async fn mongo_find_documents_core(
         projection,
         sort,
         collation,
+        None,
     )
     .await
 }
@@ -1299,6 +1300,7 @@ mod tests {
             extended_documents: Some(vec![serde_json::json!(1), serde_json::json!(2), serde_json::json!(3)]),
             total: 3,
             total_is_exact: true,
+            next_cursor: None,
         };
 
         let limited = limit_mongo_documents(result, 2);
@@ -1323,6 +1325,7 @@ mod tests {
             extended_documents: None,
             total: 1,
             total_is_exact: true,
+            next_cursor: None,
         };
 
         let query_result = mongo_show_databases_query_result(result.documents, 2).unwrap();
@@ -1348,6 +1351,7 @@ mod tests {
             extended_documents: None,
             total: 1,
             total_is_exact: true,
+            next_cursor: None,
         };
         let query_result = mongo_show_databases_query_result(empty.documents, 100).unwrap();
         assert_eq!(query_result.columns, ["name", "sizeOnDisk", "empty"]);
@@ -1360,6 +1364,7 @@ mod tests {
             extended_documents: None,
             total: 1,
             total_is_exact: true,
+            next_cursor: None,
         };
         assert!(mongo_show_databases_query_result(malformed.documents, 100).unwrap_err().contains("databases array"));
     }

crates/dbx-core/src/query.rs

@@ -2067,6 +2067,15 @@ async fn do_execute_typed(
             .map(|result| truncate_result_with_max_rows(result, max_rows))
         }
         PoolKind::HBase(_) => Err("SQL execution is not supported for HBase connections".to_string()),
+        PoolKind::DynamoDb(client) => {
+            let client = client.clone();
+            let sql = sql.to_string();
+            let max_rows = options.max_rows.unwrap_or(MAX_ROWS);
+            drop(connections);
+            // Keep the AWS SDK cold-path future off this already-large query dispatcher stack.
+            let execution = Box::pin(db::dynamodb_driver::execute_statement(&client, &sql, max_rows));
+            wait_for_query_opt(cancel_token, query_timeout, execution).await
+        }
         PoolKind::Consul(_) => Err("SQL execution is not supported for Consul connections".to_string()),
     };
     result
@@ -3498,6 +3507,7 @@ fn pool_kind_has_transactional_path(pool: &PoolKind) -> bool {
         | PoolKind::DuckDbWorker(_)
         | PoolKind::Redis(_)
         | PoolKind::MongoDb(_)
+        | PoolKind::DynamoDb(_)
         | PoolKind::Elasticsearch(_)
         | PoolKind::Easysearch(_)
         | PoolKind::Meilisearch(_)
@@ -3766,6 +3776,7 @@ pub async fn execute_statements_in_transaction_on_pool_typed(
             PoolKind::DuckDbWorker(_)
             | PoolKind::Redis(_)
             | PoolKind::MongoDb(_)
+            | PoolKind::DynamoDb(_)
             | PoolKind::Elasticsearch(_)
             | PoolKind::Easysearch(_)
             | PoolKind::Meilisearch(_)
@@ -5231,6 +5242,55 @@ for line in sys.stdin:
         }
     }
 
+    #[cfg(feature = "dynamodb")]
+    #[tokio::test]
+    #[ignore = "requires DBX_DYNAMODB_ENDPOINT and an orders table"]
+    async fn live_dynamodb_editor_scan_serializes_one_thousand_rows() {
+        let endpoint = std::env::var("DBX_DYNAMODB_ENDPOINT").expect("DBX_DYNAMODB_ENDPOINT is required");
+        let (ssl, address) = endpoint
+            .strip_prefix("https://")
+            .map(|address| (true, address))
+            .or_else(|| endpoint.strip_prefix("http://").map(|address| (false, address)))
+            .expect("DynamoDB endpoint must start with http:// or https://");
+        let (host, port) = address.rsplit_once(':').expect("DynamoDB endpoint must include a port");
+        let dir = std::env::temp_dir().join(format!("dbx-query-dynamodb-{}", uuid::Uuid::new_v4()));
+        std::fs::create_dir_all(&dir).unwrap();
+        let storage = Storage::open(&dir.join("storage.db")).await.unwrap();
+        let state = AppState::new(storage);
+        let mut config = test_connection_config(DatabaseType::DynamoDb);
+        config.host = host.to_string();
+        config.port = port.parse().expect("valid DynamoDB port");
+        config.username = "dummy".to_string();
+        config.password = "dummy".to_string();
+        config.database = Some("us-east-1".to_string());
+        config.ssl = ssl;
+        let client = db::dynamodb_driver::connect(&config, host, config.port).unwrap();
+        db::dynamodb_driver::test_connection(&client, Duration::from_secs(5)).await.unwrap();
+        state.configs.write().await.insert(config.id.clone(), config.clone());
+        state.connections.write().await.insert(config.id.clone(), PoolKind::DynamoDb(client));
+
+        let results = execute_multi_core_with_options_for_client_and_progress_typed(
+            &state,
+            &config.id,
+            "us-east-1",
+            "DBX DYNAMODB SCAN\ntable: \"orders\"\nlimit: 1000",
+            None,
+            None,
+            QueryExecutionOptions { max_rows: Some(1000), ..Default::default() },
+            None,
+        )
+        .await
+        .unwrap();
+
+        assert_eq!(results.len(), 1);
+        assert_eq!(results[0].result.rows.len(), 1000);
+        let serialized = serde_json::to_string(&results).unwrap();
+        assert!(!serialized.is_empty());
+
+        drop(state);
+        let _ = std::fs::remove_dir_all(dir);
+    }
+
     async fn agent_error_state(
         disposition: &str,
     ) -> (AppState, std::path::PathBuf, std::sync::Arc<crate::db::agent_driver::AgentRuntimeClient>) {

crates/dbx-core/src/query_execution_sql.rs

@@ -247,12 +247,27 @@ pub fn is_write_sql_for_database(sql: &str, database_type: DatabaseType) -> bool
     if database_type == DatabaseType::VictoriaMetrics {
         return false;
     }
+    if database_type == DatabaseType::DynamoDb {
+        if let Some(is_write) = classify_dynamodb_statement_write(sql) {
+            return is_write;
+        }
+    }
     if let Some(risk) = classify_search_engine_query_risk(sql, database_type) {
         return risk != SearchEngineQueryRisk::ReadOnly;
     }
     is_write_sql_with_database_type(sql, Some(database_type))
 }
 
+fn classify_dynamodb_statement_write(source: &str) -> Option<bool> {
+    let header = source.lines().find(|line| !line.trim().is_empty())?.trim().to_ascii_uppercase();
+    match header.as_str() {
+        "DBX DYNAMODB SCAN" | "DBX DYNAMODB QUERY / SCAN" => Some(false),
+        "DBX DYNAMODB INSERT ITEM" | "DBX DYNAMODB PUT ITEM" | "DBX DYNAMODB DELETE ITEM" => Some(true),
+        _ if header.starts_with("DBX DYNAMODB") => Some(true),
+        _ => None,
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub(crate) enum SearchEngineQueryRisk {
     ReadOnly,
@@ -1598,6 +1613,24 @@ mod tests {
         assert_eq!(strip_sql_comments("SELECT 'hello /* not a comment */'"), "SELECT 'hello  '");
     }
 
+    #[test]
+    fn classifies_dynamodb_generated_statements_for_read_only_protection() {
+        assert!(!is_write_sql_for_database(
+            "DBX DYNAMODB SCAN\ntable: \"orders\"\nlimit: 1000",
+            DatabaseType::DynamoDb
+        ));
+        assert!(!is_write_sql_for_database(
+            "DBX DYNAMODB QUERY / SCAN\ntable: \"orders\"\nfilter:\n{\"status\":\"SHIPPED\"}",
+            DatabaseType::DynamoDb
+        ));
+        for operation in ["INSERT ITEM", "PUT ITEM", "DELETE ITEM", "UNKNOWN"] {
+            assert!(is_write_sql_for_database(
+                &format!("DBX DYNAMODB {operation}\ntable: \"orders\""),
+                DatabaseType::DynamoDb
+            ));
+        }
+    }
+
     #[test]
     fn classifies_search_engine_rest_writes() {
         assert!(!is_write_sql_for_database("GET /_cluster/health", DatabaseType::Easysearch));

crates/dbx-core/src/sql_dialect/ddl_profile.rs

@@ -675,10 +675,9 @@ pub fn profile_for(db_type: DatabaseType) -> DdlDialectProfile {
         }
 
         // Non-tabular / not applicable for relational CREATE TABLE
-        Redis | MongoDb | Elasticsearch | Easysearch | Meilisearch | Qdrant | Milvus | Weaviate | ChromaDb | Neo4j
-        | Cassandra | Etcd | ZooKeeper | Nacos | Consul | InfluxDb | VictoriaMetrics | MessageQueue | Mqtt | Hbase => {
-            conservative_ansi(db_type)
-        }
+        Redis | MongoDb | DynamoDb | Elasticsearch | Easysearch | Meilisearch | Qdrant | Milvus | Weaviate
+        | ChromaDb | Neo4j | Cassandra | Etcd | ZooKeeper | Nacos | Consul | InfluxDb | VictoriaMetrics
+        | MessageQueue | Mqtt | Hbase => conservative_ansi(db_type),
     }
 }
 

crates/dbx-web/Cargo.toml

@@ -9,8 +9,9 @@ name = "dbx-web"
 path = "src/main.rs"
 
 [features]
-default = ["duckdb-sidecar", "mq-admin", "sqlite-sqlcipher", "system-fonts"]
+default = ["duckdb-sidecar", "dynamodb", "mq-admin", "sqlite-sqlcipher", "system-fonts"]
 duckdb-sidecar = ["dbx-core/duckdb-sidecar"]
+dynamodb = ["dbx-core/dynamodb"]
 mq-admin = ["dbx-core/mq-admin"]
 sqlite-sqlcipher = ["dbx-core/sqlite-sqlcipher"]
 system-fonts = ["dbx-core/system-fonts"]

crates/dbx-web/src/main.rs

@@ -777,6 +777,8 @@ async fn main() {
         .route("/document-store/list-databases", post(routes::document_store::list_databases))
         .route("/document-store/list-collections", post(routes::document_store::list_collections))
         .route("/document-store/find-documents", post(routes::document_store::find_documents))
+        .route("/document-store/count-documents", post(routes::document_store::count_documents))
+        .route("/document-store/dynamodb-describe-table", post(routes::document_store::describe_dynamodb_table))
         .route(
             "/document-store/elasticsearch-count-documents",
             post(routes::document_store::elasticsearch_count_documents),

crates/dbx-web/src/routes/document_store.rs

@@ -67,9 +67,26 @@ pub struct DocumentFindRequest {
     pub projection: Option<String>,
     pub sort: Option<String>,
     pub collation: Option<String>,
+    pub cursor: Option<String>,
     pub execution_id: Option<String>,
 }
 
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DocumentCountRequest {
+    pub connection_id: String,
+    pub collection: String,
+    pub filter: Option<String>,
+    pub execution_id: Option<String>,
+}
+
+#[derive(Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct DynamoDbDescribeTableRequest {
+    pub connection_id: String,
+    pub table: String,
+}
+
 #[derive(Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct ElasticsearchCountDocumentsRequest {
@@ -204,12 +221,41 @@ pub async fn find_documents(
             req.projection.as_deref(),
             req.sort.as_deref(),
             req.collation.as_deref(),
+            req.cursor.as_deref(),
         ),
     )
     .await?;
     Ok(Json(result))
 }
 
+pub async fn count_documents(
+    State(state): State<Arc<WebState>>,
+    Json(req): Json<DocumentCountRequest>,
+) -> Result<Json<u64>, AppError> {
+    let result = run_cancellable(
+        &state,
+        req.execution_id,
+        dbx_core::document_ops::count_document_store_documents_core(
+            &state.app,
+            &req.connection_id,
+            &req.collection,
+            req.filter.as_deref(),
+        ),
+    )
+    .await?;
+    Ok(Json(result))
+}
+
+pub async fn describe_dynamodb_table(
+    State(state): State<Arc<WebState>>,
+    Json(req): Json<DynamoDbDescribeTableRequest>,
+) -> Result<Json<dbx_core::db::dynamodb_driver::DynamoDbTableDescription>, AppError> {
+    let result = dbx_core::document_ops::describe_dynamodb_table_core(&state.app, &req.connection_id, &req.table)
+        .await
+        .map_err(AppError::from)?;
+    Ok(Json(result))
+}
+
 pub async fn elasticsearch_count_documents(
     State(state): State<Arc<WebState>>,
     Json(req): Json<ElasticsearchCountDocumentsRequest>,

crates/dbx-web/src/routes/mongo.rs

@@ -466,6 +466,7 @@ pub async fn find_documents(
             req.projection.as_deref(),
             req.sort.as_deref(),
             req.collation.as_deref(),
+            None,
         ),
     )
     .await?;