dataforge/data_factory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs(flows): DoD obligatorio con user-facing surface + abrir issues 0100-0103 (taxonomia, frontmatter migration, dev_console, work dashboard)

Browse Source 
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Egutierrez
2026-05-17 00:07:04 +02:00
parent 6337d93491
commit 8c5152fca4
12 changed files with 765 additions and 170 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CMakeLists.txt
+2 -4
View File
@@ -9,10 +9,8 @@ add_imgui_app(data_factory
)
target_include_directories(data_factory PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
# fn_table_viz: optional, kept for parity with dag_engine_ui.
if(TARGET fn_table_viz)
target_link_libraries(data_factory PRIVATE fn_table_viz)
endif()
# fn_module_data_table: required — tabs.cpp uses data_table::render (issue 0081).
target_link_libraries(data_factory PRIVATE fn_module_data_table)
if(WIN32)
target_link_libraries(data_factory PRIVATE ws2_32)
app.md
+5
View File
@@ -7,11 +7,16 @@ tags: [imgui, dashboard, data-pipeline, factory, http, websocket]
uses_functions:
- empty_state_cpp_core
- badge_cpp_core
- data_table_cpp_viz
uses_types: []
uses_modules: [data_table_cpp]
framework: "imgui"
entry_point: "main.cpp"
dir_path: "apps/data_factory"
repo_url: "https://gitea.organic-machine.com/dataforge/data_factory"
icon:
phosphor: "factory"
accent: "#f97316"
e2e_checks:
- id: build_cmake
cmd: "cmake --build cpp/build -j --target data_factory"
data/hn_top_stories.duckdb
BIN
View File
Binary file not shown.
data_factory.db-shm
BIN
View File
Binary file not shown.
data_factory.db-wal
BIN
View File
Binary file not shown.
data_http.cpp
+92
View File
@@ -102,6 +102,17 @@ static void parse_run(const json& j, Run& r) {
r.duration_ms = get_int64(j, "duration_ms");
r.trigger = get_str(j, "trigger");
r.error = get_str(j, "error");
r.storage_db_id = get_str(j, "storage_db_id");
r.storage_table = get_str(j, "storage_table");
}
static void parse_table_entry(const json& j, TableEntry& t) {
t.database_id = get_str(j, "database_id");
t.database_label = get_str(j, "database_label");
t.database_kind = get_str(j, "database_kind");
t.table_name = get_str(j, "table_name");
t.row_count = get_int64(j, "row_count");
t.error = get_str(j, "error");
}
static void parse_db(const json& j, DatabaseInfo& d) {
@@ -191,6 +202,30 @@ bool list_databases_http(const std::string& api_url,
return true;
}
bool list_tables_http(const std::string& api_url,
std::vector<TableEntry>& out) {
std::string host;
int port;
if (!parse_url(api_url, host, port)) return false;
HttpClient cli(host, port);
auto res = cli.get("/api/datafactory/tables");
if (!res.ok()) {
fprintf(stderr, "[df_http] list_tables failed: status=%d\n", res.status);
return false;
}
auto j = json::parse(res.body, nullptr, false);
if (!j.is_object() || !j.contains("tables") || !j["tables"].is_array()) {
return false;
}
out.clear();
for (auto& item : j["tables"]) {
TableEntry t;
parse_table_entry(item, t);
out.push_back(std::move(t));
}
return true;
}
bool get_function_http(const std::string& api_url,
const std::string& function_id,
FnInfo& out) {
@@ -219,4 +254,61 @@ bool get_function_http(const std::string& api_url,
return true;
}
bool get_table_preview_http(const std::string& api_url,
const std::string& database_id,
const std::string& table,
int limit, int offset,
TablePreview& out) {
std::string host;
int port;
if (!parse_url(api_url, host, port)) return false;
if (database_id.empty() || table.empty()) return false;
HttpClient cli(host, port);
// Build query string manually (no URL encoding needed — IDs/table names
// are alphanumeric per the server-side validation regex).
char path[512];
std::snprintf(path, sizeof(path),
"/api/datafactory/preview?database_id=%s&table=%s&limit=%d&offset=%d",
database_id.c_str(), table.c_str(), limit, offset);
auto res = cli.get(path);
if (!res.ok()) {
fprintf(stderr, "[df_http] get_table_preview(%s.%s) failed: status=%d body=%s\n",
database_id.c_str(), table.c_str(), res.status, res.body.c_str());
return false;
}
auto j = json::parse(res.body, nullptr, false);
if (!j.is_object()) return false;
out.database_id = get_str(j, "database_id");
out.table_name = get_str(j, "table_name");
out.total_rows = get_int64(j, "total_rows");
out.limit = get_int64(j, "limit");
out.offset = get_int64(j, "offset");
out.columns.clear();
if (j.contains("columns") && j["columns"].is_array()) {
for (auto& c : j["columns"]) {
std::string name = get_str(c, "name");
std::string type = get_str(c, "type");
out.columns.emplace_back(name, type);
}
}
out.rows.clear();
if (j.contains("rows") && j["rows"].is_array()) {
for (auto& row : j["rows"]) {
if (!row.is_array()) continue;
std::vector<std::string> r;
r.reserve(row.size());
for (auto& cell : row) {
if (cell.is_string()) r.push_back(cell.get<std::string>());
else if (cell.is_null()) r.push_back("");
else r.push_back(cell.dump());
}
out.rows.push_back(std::move(r));
}
}
return true;
}
} // namespace data_factory
data_http.h
+30
View File
@@ -34,6 +34,8 @@ struct Run {
long long duration_ms = 0;
std::string trigger;
std::string error;
std::string storage_db_id;
std::string storage_table;
};
struct DatabaseInfo {
@@ -47,6 +49,25 @@ struct DatabaseInfo {
std::string last_seen_at;
};
struct TableEntry {
std::string database_id;
std::string database_label;
std::string database_kind;
std::string table_name;
long long row_count = 0;
std::string error;
};
struct TablePreview {
std::string database_id;
std::string table_name;
std::vector<std::pair<std::string, std::string>> columns; // (name, type)
std::vector<std::vector<std::string>> rows;
long long total_rows = 0;
long long limit = 100;
long long offset = 0;
};
// Mirrors dag_engine_ui FnInfo (response shape of GET /api/functions/{id}).
struct FnInfo {
std::string id;
@@ -69,8 +90,17 @@ bool list_runs_http(const std::string& api_url, const std::string& node_id,
bool list_databases_http(const std::string& api_url,
std::vector<DatabaseInfo>& out);
bool list_tables_http(const std::string& api_url,
std::vector<TableEntry>& out);
bool get_function_http(const std::string& api_url,
const std::string& function_id,
FnInfo& out);
bool get_table_preview_http(const std::string& api_url,
const std::string& database_id,
const std::string& table,
int limit, int offset,
TablePreview& out);
} // namespace data_factory
main.cpp
+9
View File
@@ -26,6 +26,7 @@ static std::string g_ws_path = "/api/ws/datafactory";
static std::vector<data_factory::Node> g_nodes;
static std::vector<data_factory::Run> g_runs_all;
static std::vector<data_factory::DatabaseInfo> g_databases;
static std::vector<data_factory::TableEntry> g_tables;
static WsClient g_ws;
static int g_ws_msg_count = 0;
static bool g_initial_fetched = false;
@@ -36,9 +37,11 @@ static bool g_show_map = true;
static bool g_show_extractors = true;
static bool g_show_transformers = true;
static bool g_show_databases = true;
static bool g_show_tables = true;
static bool g_show_sinks = true;
static bool g_show_health = true;
static bool g_show_detail = true;
static bool g_show_table_preview = true;
static bool g_show_live = false;
static void upsert_run(const data_factory::Run& r) {
@@ -132,6 +135,7 @@ static void render() {
g_refresh_pending = false;
data_factory::list_nodes_http(g_api_url, "", g_nodes);
data_factory::list_databases_http(g_api_url, g_databases);
data_factory::list_tables_http(g_api_url, g_tables);
std::vector<data_factory::Run> tmp;
if (data_factory::list_runs_http(g_api_url, "", 200, tmp)) {
for (auto& r : tmp) upsert_run(r);
@@ -149,10 +153,13 @@ static void render() {
if (g_show_extractors) data_factory_ui::draw_extractors(g_api_url, g_nodes, g_runs_all);
if (g_show_transformers) data_factory_ui::draw_transformers(g_api_url, g_nodes, g_runs_all);
if (g_show_databases) data_factory_ui::draw_databases(g_api_url, g_databases);
if (g_show_tables) data_factory_ui::draw_tables(g_api_url, g_tables);
if (g_show_sinks) data_factory_ui::draw_sinks(g_api_url, g_nodes, g_runs_all);
if (g_show_health) data_factory_ui::draw_health(g_api_url, g_runs_all);
if (g_show_detail) data_factory_ui::draw_node_detail_panel(
g_api_url, g_nodes, g_runs_all, &g_show_detail);
if (g_show_table_preview) data_factory_ui::draw_table_preview_panel(
g_api_url, &g_show_table_preview);
if (g_show_live) draw_live();
}
@@ -183,6 +190,8 @@ int main(int argc, char** argv) {
{ "Extractors", nullptr, &g_show_extractors },
{ "Transformers", nullptr, &g_show_transformers },
{ "Databases", nullptr, &g_show_databases },
{ "Tables", nullptr, &g_show_tables },
{ "Table Preview", nullptr, &g_show_table_preview },
{ "Sinks", nullptr, &g_show_sinks },
{ "Health", nullptr, &g_show_health },
{ "Node Detail", nullptr, &g_show_detail },
migrations/002_add_storage_columns.sql
+5
View File
@@ -0,0 +1,5 @@
-- Migration 002: track where each run's extracted data is stored.
-- Aditiva, idempotente (SQLite ALTER ADD COLUMN + "duplicate column" ignorado en app code).
ALTER TABLE runs ADD COLUMN storage_db_id TEXT NOT NULL DEFAULT '';
ALTER TABLE runs ADD COLUMN storage_table TEXT NOT NULL DEFAULT '';
operations.db
View File
tabs.cpp
+595 -144
View File
@@ -1,4 +1,6 @@
#include "tabs.h"
#include "data_table/data_table.h"
#include "core/data_table_types.h"
#include "core/icons_tabler.h"
#include "core/empty_state.h"
#include "core/badge.h"
@@ -9,9 +11,115 @@
#include <ctime>
#include <map>
#include <string>
#include <vector>
namespace data_factory_ui {
// data_table::State persistente por panel (issue 0081 pattern).
static data_table::State g_st_tables;
static data_table::State g_st_nodes_extractors;
static data_table::State g_st_nodes_transformers;
static data_table::State g_st_nodes_sinks;
static data_table::State g_st_databases;
static data_table::State g_st_kpis;
static data_table::State g_st_node_runs;
static data_table::State g_st_preview;
// Backing storage for cell strings (owns chars referenced by TableInput.cells).
static std::vector<std::string> g_back_extractors;
static std::vector<const char*> g_ptrs_extractors;
static std::vector<std::string> g_back_transformers;
static std::vector<const char*> g_ptrs_transformers;
static std::vector<std::string> g_back_sinks;
static std::vector<const char*> g_ptrs_sinks;
static std::vector<std::string> g_back_tables;
static std::vector<const char*> g_ptrs_tables;
static std::vector<std::string> g_back_databases;
static std::vector<const char*> g_ptrs_databases;
static std::vector<std::string> g_back_kpis;
static std::vector<const char*> g_ptrs_kpis;
static std::vector<std::string> g_back_node_runs;
static std::vector<const char*> g_ptrs_node_runs;
static std::vector<std::string> g_back_preview;
static std::vector<const char*> g_ptrs_preview;
// ---------------------------------------------------------------------------
// Table preview state (populated by double-click in draw_tables).
// ---------------------------------------------------------------------------
struct PreviewSelection {
std::string database_id;
std::string table_name;
data_factory::TablePreview cache;
bool loading = false;
bool loaded = false;
std::string error;
int offset = 0;
int limit = 100;
};
static PreviewSelection& preview_state() {
static PreviewSelection s;
return s;
}
static void cells_to_ptrs(const std::vector<std::string>& backing,
std::vector<const char*>& ptrs) {
ptrs.resize(backing.size());
for (size_t i = 0; i < backing.size(); i++) ptrs[i] = backing[i].c_str();
}
// Shared BadgeRules.
static std::vector<data_table::BadgeRule> run_status_badges() {
return {
{"success", "#22c55e", "success"},
{"failed", "#ef4444", "failed"},
{"running", "#3b82f6", "running"},
{"pending", "#94a3b8", "pending"},
{"cancelled", "#6b7280", "cancelled"},
};
}
static std::vector<data_table::BadgeRule> kind_badges() {
return {
{"extractor", "#0ea5e9", "extractor"},
{"transformer", "#a855f7", "transformer"},
{"sink", "#f97316", "sink"},
{"database", "#14b8a6", "database"},
{"validator", "#eab308", "validator"},
};
}
static std::vector<data_table::BadgeRule> enabled_badges() {
return {
{"yes", "#22c55e", "yes"},
{"no", "#6b7280", "no"},
};
}
// CategoricalChip helpers (dot izquierda + texto, siempre visible).
static std::vector<data_table::ChipRule> run_status_chips() {
return {
{"success", "#22c55e"},
{"failed", "#ef4444"},
{"running", "#3b82f6"},
{"pending", "#94a3b8"},
{"cancelled", "#6b7280"},
};
}
static std::vector<data_table::ChipRule> kind_chips() {
return {
{"extractor", "#0ea5e9"},
{"transformer", "#a855f7"},
{"sink", "#f97316"},
{"database", "#14b8a6"},
{"validator", "#eab308"},
{"duckdb", "#f59e0b"},
{"sqlite", "#6366f1"},
};
}
// ---------------------------------------------------------------------------
// Globals
// ---------------------------------------------------------------------------
@@ -64,14 +172,6 @@ static time_t parse_rfc3339(const std::string& s) {
return std::mktime(&tm);
}
static BadgeVariant variant_for_status(const std::string& st) {
if (st == "success") return BadgeVariant::Success;
if (st == "failed") return BadgeVariant::Error;
if (st == "running") return BadgeVariant::Warning;
if (st == "cancelled") return BadgeVariant::Default;
return BadgeVariant::Default;
}
// Pick most-recent run per node from runs_all.
static const data_factory::Run* last_run_for(
const std::string& node_id,
@@ -89,83 +189,119 @@ static const data_factory::Run* last_run_for(
// Generic kind table — used by extractors / transformers / sinks
// ---------------------------------------------------------------------------
static void draw_node_table(const char* table_id,
// Render the node table for a given kind via data_table::render.
// `kind_label` selects which static State/backing to use (one per kind so the
// three calls don't clobber each other's sort/filter/breadcrumb state).
static void draw_node_table(const char* /*table_id*/,
const std::vector<data_factory::Node>& nodes,
const std::string& filter_kind,
const std::vector<data_factory::Run>& runs_all,
bool show_schedule)
{
int cols = show_schedule ? 6 : 5;
if (!ImGui::BeginTable(table_id, cols,
ImGuiTableFlags_Borders | ImGuiTableFlags_RowBg |
ImGuiTableFlags_Resizable | ImGuiTableFlags_ScrollY))
{
return;
// Pick the per-kind state + backing.
data_table::State* st = &g_st_nodes_extractors;
std::vector<std::string>* backing = &g_back_extractors;
std::vector<const char*>* ptrs = &g_ptrs_extractors;
const char* dt_id = "##dt_extractors";
if (filter_kind == "transformer") {
st = &g_st_nodes_transformers;
backing = &g_back_transformers;
ptrs = &g_ptrs_transformers;
dt_id = "##dt_transformers";
} else if (filter_kind == "sink") {
st = &g_st_nodes_sinks;
backing = &g_back_sinks;
ptrs = &g_ptrs_sinks;
dt_id = "##dt_sinks";
}
ImGui::TableSetupColumn("Name", ImGuiTableColumnFlags_WidthStretch, 0.20f);
ImGui::TableSetupColumn("Function", ImGuiTableColumnFlags_WidthStretch, 0.30f);
// Filter nodes for the current kind, and pre-resolve their last-run.
std::vector<const data_factory::Node*> filtered;
filtered.reserve(nodes.size());
for (auto& n : nodes) if (n.kind == filter_kind) filtered.push_back(&n);
data_table::TableInput tbl;
tbl.name = filter_kind;
if (show_schedule) {
tbl.headers = {"Name", "Function", "Schedule", "Last Run", "Status", "Rows/KB", "Enabled"};
tbl.types = {
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
};
} else {
tbl.headers = {"Name", "Function", "Last Run", "Status", "Rows/KB", "Enabled"};
tbl.types = {
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
};
}
tbl.rows = static_cast<int>(filtered.size());
tbl.cols = static_cast<int>(tbl.headers.size());
tbl.column_specs.resize(tbl.cols);
for (int i = 0; i < tbl.cols; i++) tbl.column_specs[i].id = tbl.headers[i];
// Function column (Code renderer not in v1.3.x; keep Text — function_id is plain).
// Status column (CategoricalChip — dot izquierda + texto, siempre visible).
int status_col = show_schedule ? 4 : 3;
tbl.column_specs[status_col].renderer = data_table::CellRenderer::CategoricalChip;
tbl.column_specs[status_col].chips = run_status_chips();
// Enabled column (CategoricalChip yes/no).
int enabled_col = tbl.cols - 1;
tbl.column_specs[enabled_col].renderer = data_table::CellRenderer::CategoricalChip;
tbl.column_specs[enabled_col].chips = {
{"yes", "#22c55e"},
{"no", "#6b7280"},
};
backing->clear();
backing->reserve(filtered.size() * tbl.cols);
for (auto* pn : filtered) {
const data_factory::Node& n = *pn;
const data_factory::Run* lr = last_run_for(n.id, runs_all);
backing->push_back(n.name);
backing->push_back(n.function_id.empty() ? "(none)" : n.function_id);
if (show_schedule)
ImGui::TableSetupColumn("Schedule", ImGuiTableColumnFlags_WidthStretch, 0.12f);
ImGui::TableSetupColumn("Last Run", ImGuiTableColumnFlags_WidthStretch, 0.18f);
ImGui::TableSetupColumn("Status", ImGuiTableColumnFlags_WidthStretch, 0.10f);
ImGui::TableSetupColumn("Rows/KB", ImGuiTableColumnFlags_WidthStretch, 0.10f);
ImGui::TableHeadersRow();
backing->push_back(n.schedule_cron.empty() ? "manual" : n.schedule_cron);
backing->push_back(lr ? lr->started_at : "-");
backing->push_back(lr ? lr->status : "-");
if (lr) {
char buf[64];
std::snprintf(buf, sizeof(buf), "%lld / %lld", lr->rows_out, lr->kb_out);
backing->push_back(buf);
} else {
backing->push_back("-");
}
backing->push_back(n.enabled ? "yes" : "no");
}
cells_to_ptrs(*backing, *ptrs);
tbl.cells = ptrs->data();
int row_idx = 0;
for (auto& n : nodes) {
if (n.kind != filter_kind) continue;
ImGui::PushID(row_idx++);
ImGui::TableNextRow();
std::vector<data_table::TableEvent> events;
data_table::render(dt_id, {tbl}, *st, &events);
// Name (selectable -> sets selection)
ImGui::TableNextColumn();
bool selected = (selection().node_id == n.id);
if (ImGui::Selectable(n.name.c_str(), selected,
ImGuiSelectableFlags_SpanAllColumns))
for (auto& ev : events) {
if (ev.kind == data_table::TableEventKind::RowDoubleClick &&
ev.row >= 0 && ev.row < static_cast<int>(filtered.size()))
{
const data_factory::Node& n = *filtered[ev.row];
selection().node_id = n.id;
// Invalidate function cache if function_id changed.
if (function_cache().function_id != n.function_id) {
function_cache() = {};
function_cache().function_id = n.function_id;
}
}
if (!n.enabled) {
ImGui::SameLine();
badge("disabled", BadgeVariant::Default);
}
// Function id
ImGui::TableNextColumn();
if (n.function_id.empty()) ImGui::TextDisabled("(none)");
else ImGui::TextUnformatted(n.function_id.c_str());
// Schedule
if (show_schedule) {
ImGui::TableNextColumn();
if (n.schedule_cron.empty()) ImGui::TextDisabled("manual");
else ImGui::TextUnformatted(n.schedule_cron.c_str());
}
// Last run
const data_factory::Run* lr = last_run_for(n.id, runs_all);
ImGui::TableNextColumn();
if (lr) ImGui::TextUnformatted(lr->started_at.c_str());
else ImGui::TextDisabled("-");
// Status badge
ImGui::TableNextColumn();
if (lr) badge(lr->status.c_str(), variant_for_status(lr->status));
else ImGui::TextDisabled("-");
// Rows / KB
ImGui::TableNextColumn();
if (lr) ImGui::Text("%lld / %lld", lr->rows_out, lr->kb_out);
else ImGui::TextDisabled("-");
ImGui::PopID();
}
ImGui::EndTable();
}
// ---------------------------------------------------------------------------
@@ -243,6 +379,101 @@ void draw_sinks(const std::string& /*api_url*/,
ImGui::End();
}
// ---------------------------------------------------------------------------
// Tables
// ---------------------------------------------------------------------------
void draw_tables(const std::string& /*api_url*/,
const std::vector<data_factory::TableEntry>& tables)
{
if (!ImGui::Begin(TI_TABLE " Tables")) {
ImGui::End();
return;
}
if (tables.empty()) {
empty_state(TI_TABLE, "No tables found",
"Register databases in data_factory.db to see their tables here.");
ImGui::End();
return;
}
// Count errors separately.
int error_count = 0;
for (auto& t : tables) if (!t.error.empty()) error_count++;
if (error_count > 0) {
ImGui::TextDisabled("%zu tables across all databases (%d error(s)).",
tables.size(), error_count);
} else {
ImGui::TextDisabled("%zu tables across all databases.", tables.size());
}
{
data_table::TableInput tbl;
tbl.name = "tables";
tbl.headers = {"Database", "Kind", "Table", "Rows"};
tbl.types = {
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::Int,
};
tbl.rows = static_cast<int>(tables.size());
tbl.cols = static_cast<int>(tbl.headers.size());
tbl.column_specs.resize(tbl.cols);
for (int i = 0; i < tbl.cols; i++) tbl.column_specs[i].id = tbl.headers[i];
// Kind column: CategoricalChip (duckdb = amber, sqlite = indigo).
tbl.column_specs[1].renderer = data_table::CellRenderer::CategoricalChip;
tbl.column_specs[1].chips = kind_chips();
// Rows column: ColorScale (0 = dim, up to 10000 = bright).
tbl.column_specs[3].renderer = data_table::CellRenderer::ColorScale;
tbl.column_specs[3].range_min = 0.0;
tbl.column_specs[3].range_max = 10000.0;
tbl.column_specs[3].range_alpha = 0.25f;
g_back_tables.clear();
g_back_tables.reserve(tables.size() * tbl.cols);
for (auto& t : tables) {
g_back_tables.push_back(t.database_label.empty() ? t.database_id : t.database_label);
g_back_tables.push_back(t.database_kind);
// Show table name; if error, show the error text in the table cell.
g_back_tables.push_back(t.error.empty() ? t.table_name
: t.table_name + " — " + t.error);
{
char buf[32];
std::snprintf(buf, sizeof(buf), "%lld", t.row_count);
g_back_tables.push_back(buf);
}
}
cells_to_ptrs(g_back_tables, g_ptrs_tables);
tbl.cells = g_ptrs_tables.data();
std::vector<data_table::TableEvent> tbl_events;
data_table::render("##dt_tables", {tbl}, g_st_tables, &tbl_events);
for (auto& ev : tbl_events) {
if (ev.kind == data_table::TableEventKind::RowDoubleClick &&
ev.row >= 0 && ev.row < static_cast<int>(tables.size()))
{
const auto& t = tables[ev.row];
auto& ps = preview_state();
// Only reset if selection changed.
if (ps.database_id != t.database_id || ps.table_name != t.table_name) {
ps.database_id = t.database_id;
ps.table_name = t.table_name;
ps.loaded = false;
ps.loading = false;
ps.error.clear();
ps.offset = 0;
ps.cache = data_factory::TablePreview{};
}
}
}
}
ImGui::End();
}
// ---------------------------------------------------------------------------
// Databases
// ---------------------------------------------------------------------------
@@ -260,35 +491,44 @@ void draw_databases(const std::string& /*api_url*/,
ImGui::End();
return;
}
if (ImGui::BeginTable("##df_databases", 6,
ImGuiTableFlags_Borders | ImGuiTableFlags_RowBg |
ImGuiTableFlags_Resizable | ImGuiTableFlags_ScrollY))
{
ImGui::TableSetupColumn("Label", ImGuiTableColumnFlags_WidthStretch, 0.18f);
ImGui::TableSetupColumn("Kind", ImGuiTableColumnFlags_WidthStretch, 0.10f);
ImGui::TableSetupColumn("URI", ImGuiTableColumnFlags_WidthStretch, 0.32f);
ImGui::TableSetupColumn("Tables", ImGuiTableColumnFlags_WidthStretch, 0.10f);
ImGui::TableSetupColumn("Size", ImGuiTableColumnFlags_WidthStretch, 0.15f);
ImGui::TableSetupColumn("Last Seen",ImGuiTableColumnFlags_WidthStretch, 0.15f);
ImGui::TableHeadersRow();
data_table::TableInput tbl;
tbl.name = "databases";
tbl.headers = {"Label", "Kind", "URI", "Tables", "Size", "Last Seen"};
tbl.types = {
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::Int,
data_table::ColumnType::String,
data_table::ColumnType::String,
};
tbl.rows = static_cast<int>(dbs.size());
tbl.cols = static_cast<int>(tbl.headers.size());
tbl.column_specs.resize(tbl.cols);
for (int i = 0; i < tbl.cols; i++) tbl.column_specs[i].id = tbl.headers[i];
tbl.column_specs[1].renderer = data_table::CellRenderer::CategoricalChip;
tbl.column_specs[1].chips = kind_chips();
g_back_databases.clear();
g_back_databases.reserve(dbs.size() * tbl.cols);
for (auto& d : dbs) {
ImGui::TableNextRow();
ImGui::TableNextColumn();
ImGui::TextUnformatted(d.label.empty() ? d.id.c_str() : d.label.c_str());
ImGui::TableNextColumn();
badge(d.kind.c_str(), BadgeVariant::Info);
ImGui::TableNextColumn();
ImGui::TextUnformatted(d.uri.c_str());
ImGui::TableNextColumn();
ImGui::Text("%lld", d.table_count);
ImGui::TableNextColumn();
ImGui::TextUnformatted(format_bytes(d.size_bytes).c_str());
ImGui::TableNextColumn();
if (d.last_seen_at.empty()) ImGui::TextDisabled("-");
else ImGui::TextUnformatted(d.last_seen_at.c_str());
g_back_databases.push_back(d.label.empty() ? d.id : d.label);
g_back_databases.push_back(d.kind);
g_back_databases.push_back(d.uri);
{
char buf[32];
std::snprintf(buf, sizeof(buf), "%lld", d.table_count);
g_back_databases.push_back(buf);
}
ImGui::EndTable();
g_back_databases.push_back(format_bytes(d.size_bytes));
g_back_databases.push_back(d.last_seen_at.empty() ? "-" : d.last_seen_at);
}
cells_to_ptrs(g_back_databases, g_ptrs_databases);
tbl.cells = g_ptrs_databases.data();
data_table::render("##dt_databases", {tbl}, g_st_databases);
}
ImGui::End();
}
@@ -338,35 +578,45 @@ void draw_health(const std::string& /*api_url*/,
float success_rate = (terminal > 0)
? (100.0f * (float)success_all / (float)terminal) : 0.0f;
if (ImGui::BeginTable("##df_kpis", 4,
ImGuiTableFlags_Borders | ImGuiTableFlags_SizingStretchSame))
{
ImGui::TableNextRow();
data_table::TableInput tbl;
tbl.name = "kpis";
tbl.headers = {"KPI", "Value", "Detail"};
tbl.types = {
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
};
tbl.rows = 4;
tbl.cols = 3;
tbl.column_specs.resize(tbl.cols);
for (int i = 0; i < tbl.cols; i++) tbl.column_specs[i].id = tbl.headers[i];
ImGui::TableNextColumn();
ImGui::Text("%s Runs (24h)", TI_ACTIVITY);
ImGui::Text("%d", runs_24h);
ImGui::TextDisabled("success: %d", success_24h);
g_back_kpis.clear();
g_back_kpis.reserve(tbl.rows * tbl.cols);
char buf[64];
// Runs (24h)
g_back_kpis.push_back("Runs (24h)");
std::snprintf(buf, sizeof(buf), "%d", runs_24h); g_back_kpis.push_back(buf);
std::snprintf(buf, sizeof(buf), "success: %d", success_24h); g_back_kpis.push_back(buf);
// Success rate
g_back_kpis.push_back("Success rate");
std::snprintf(buf, sizeof(buf), "%.1f%%", success_rate); g_back_kpis.push_back(buf);
std::snprintf(buf, sizeof(buf), "%d / %d terminal", success_all, terminal);
g_back_kpis.push_back(buf);
// Failed (24h)
g_back_kpis.push_back("Failed (24h)");
std::snprintf(buf, sizeof(buf), "%d", failed_24h); g_back_kpis.push_back(buf);
std::snprintf(buf, sizeof(buf), "pending: %d", pending_total); g_back_kpis.push_back(buf);
// Throughput (24h)
g_back_kpis.push_back("Throughput (24h)");
std::snprintf(buf, sizeof(buf), "%lld rows", rows_24h); g_back_kpis.push_back(buf);
std::snprintf(buf, sizeof(buf), "%lld KB", kb_24h); g_back_kpis.push_back(buf);
ImGui::TableNextColumn();
ImGui::Text("%s Success rate", TI_CHECK);
ImGui::TextColored(ImVec4(0.30f, 0.85f, 0.40f, 1), "%.1f%%", success_rate);
ImGui::TextDisabled("%d / %d terminal", success_all, terminal);
cells_to_ptrs(g_back_kpis, g_ptrs_kpis);
tbl.cells = g_ptrs_kpis.data();
ImGui::TableNextColumn();
ImGui::Text("%s Failed (24h)", TI_ALERT_TRIANGLE);
if (failed_24h > 0)
ImGui::TextColored(ImVec4(0.95f, 0.35f, 0.30f, 1), "%d", failed_24h);
else
ImGui::Text("%d", failed_24h);
ImGui::TextDisabled("pending: %d", pending_total);
ImGui::TableNextColumn();
ImGui::Text("%s Throughput (24h)", TI_BOLT);
ImGui::Text("%lld rows", rows_24h);
ImGui::TextDisabled("%lld KB", kb_24h);
ImGui::EndTable();
data_table::render("##dt_kpis", {tbl}, g_st_kpis);
}
ImGui::Separator();
@@ -498,6 +748,27 @@ void draw_node_detail_panel(const std::string& api_url,
}
}
// Storage info: derive from most recent run with storage populated.
{
const data_factory::Run* latest_with_storage = nullptr;
for (auto& r : runs_all) {
if (r.node_id != nid) continue;
if (r.storage_db_id.empty() && r.storage_table.empty()) continue;
latest_with_storage = &r;
break;
}
if (latest_with_storage) {
ImGui::Separator();
ImGui::Text("%s Storage", TI_DATABASE);
ImGui::SameLine();
badge(latest_with_storage->storage_db_id.c_str(), BadgeVariant::Info);
ImGui::SameLine();
ImGui::TextDisabled("table:");
ImGui::SameLine();
badge(latest_with_storage->storage_table.c_str(), BadgeVariant::Default);
}
}
// Function metadata card (lazy load)
if (!node->function_id.empty()) {
ImGui::Separator();
@@ -545,37 +816,217 @@ void draw_node_detail_panel(const std::string& api_url,
// Recent runs (top 10)
ImGui::Separator();
ImGui::Text("%s Recent runs", TI_HISTORY);
int shown = 0;
if (ImGui::BeginTable("##df_node_runs", 5,
ImGuiTableFlags_Borders | ImGuiTableFlags_RowBg))
{
ImGui::TableSetupColumn("Started", ImGuiTableColumnFlags_WidthStretch, 0.30f);
ImGui::TableSetupColumn("Status", ImGuiTableColumnFlags_WidthStretch, 0.12f);
ImGui::TableSetupColumn("Duration", ImGuiTableColumnFlags_WidthStretch, 0.13f);
ImGui::TableSetupColumn("Rows", ImGuiTableColumnFlags_WidthStretch, 0.10f);
ImGui::TableSetupColumn("Trigger", ImGuiTableColumnFlags_WidthStretch, 0.15f);
ImGui::TableHeadersRow();
// Filter + cap.
std::vector<const data_factory::Run*> shown_runs;
shown_runs.reserve(10);
for (auto& r : runs_all) {
if (r.node_id != nid) continue;
if (shown >= 10) break;
shown++;
ImGui::TableNextRow();
ImGui::TableNextColumn();
ImGui::TextUnformatted(r.started_at.c_str());
ImGui::TableNextColumn();
badge(r.status.c_str(), variant_for_status(r.status));
ImGui::TableNextColumn();
ImGui::TextUnformatted(format_duration(r.duration_ms).c_str());
ImGui::TableNextColumn();
ImGui::Text("%lld", r.rows_out);
ImGui::TableNextColumn();
ImGui::TextUnformatted(r.trigger.c_str());
shown_runs.push_back(&r);
if (shown_runs.size() >= 10) break;
}
ImGui::EndTable();
}
if (shown == 0) {
if (shown_runs.empty()) {
ImGui::TextDisabled("(no runs for this node yet)");
} else {
data_table::TableInput tbl;
tbl.name = "node_runs";
tbl.headers = {"Started", "Status", "Duration (ms)", "Rows", "Trigger", "Storage DB", "Table"};
tbl.types = {
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::Float,
data_table::ColumnType::Int,
data_table::ColumnType::String,
data_table::ColumnType::String,
data_table::ColumnType::String,
};
tbl.rows = static_cast<int>(shown_runs.size());
tbl.cols = 7;
tbl.column_specs.resize(tbl.cols);
for (int i = 0; i < tbl.cols; i++) tbl.column_specs[i].id = tbl.headers[i];
tbl.column_specs[1].renderer = data_table::CellRenderer::CategoricalChip;
tbl.column_specs[1].chips = run_status_chips();
tbl.column_specs[2].renderer = data_table::CellRenderer::ColorScale;
tbl.column_specs[2].range_min = 0.0;
tbl.column_specs[2].range_max = 5000.0;
tbl.column_specs[2].range_alpha = 0.30f;
// Default 3-stop green→amber→red usado si range_stops vacio (helper interno).
// Mantenemos Duration badges semanticos en el viejo path? — no, ColorScale tinta fondo.
tbl.column_specs[2].duration_warn_ms = 1000.0f;
tbl.column_specs[2].duration_error_ms = 5000.0f;
g_back_node_runs.clear();
g_back_node_runs.reserve(shown_runs.size() * tbl.cols);
for (auto* pr : shown_runs) {
const data_factory::Run& r = *pr;
g_back_node_runs.push_back(r.started_at);
g_back_node_runs.push_back(r.status);
{
char buf[32];
std::snprintf(buf, sizeof(buf), "%lld", r.duration_ms);
g_back_node_runs.push_back(buf);
}
{
char buf[32];
std::snprintf(buf, sizeof(buf), "%lld", r.rows_out);
g_back_node_runs.push_back(buf);
}
g_back_node_runs.push_back(r.trigger);
g_back_node_runs.push_back(r.storage_db_id);
g_back_node_runs.push_back(r.storage_table);
}
cells_to_ptrs(g_back_node_runs, g_ptrs_node_runs);
tbl.cells = g_ptrs_node_runs.data();
std::vector<data_table::TableEvent> events;
data_table::render("##dt_node_runs", {tbl}, g_st_node_runs, &events);
for (auto& ev : events) {
if (ev.kind == data_table::TableEventKind::RowDoubleClick &&
ev.row >= 0 && ev.row < static_cast<int>(shown_runs.size()))
{
selection().run_id = shown_runs[ev.row]->id;
}
}
}
ImGui::End();
}
// ---------------------------------------------------------------------------
// Table preview panel (opened by double-click in Tables tab)
// ---------------------------------------------------------------------------
void draw_table_preview_panel(const std::string& api_url, bool* p_open) {
if (!ImGui::Begin(TI_EYE " Table Preview", p_open)) {
ImGui::End();
return;
}
auto& ps = preview_state();
if (ps.database_id.empty()) {
empty_state(TI_EYE, "No table selected",
"Double-click a row in the Tables tab to preview its data.");
ImGui::End();
return;
}
// Lazy fetch: trigger blocking HTTP fetch on the render thread.
// This is KISS — data_factory previews are used interactively; the one-off
// blocking call completes in <200ms over loopback.
if (!ps.loaded && !ps.loading && ps.error.empty()) {
ps.loading = true;
bool ok = data_factory::get_table_preview_http(
api_url, ps.database_id, ps.table_name, ps.limit, ps.offset, ps.cache);
ps.loading = false;
if (ok) {
ps.loaded = true;
} else {
ps.error = "Failed to fetch preview for " + ps.database_id + "." + ps.table_name;
}
}
// Header: "database_id . table_name (total_rows rows)"
{
char hdr[256];
long long total = ps.loaded ? ps.cache.total_rows : 0;
std::snprintf(hdr, sizeof(hdr), "%s %s (%lld rows)",
ps.database_id.c_str(), ps.table_name.c_str(), total);
ImGui::Text("%s %s", TI_TABLE, hdr);
}
ImGui::Separator();
if (!ps.loaded && ps.error.empty()) {
ImGui::TextDisabled("Loading...");
ImGui::End();
return;
}
if (!ps.error.empty()) {
ImGui::TextColored(ImVec4(0.95f, 0.4f, 0.4f, 1), "%s", ps.error.c_str());
ImGui::End();
return;
}
// Schema summary.
if (!ps.cache.columns.empty()) {
std::string schema_line;
for (size_t i = 0; i < ps.cache.columns.size(); i++) {
if (i > 0) schema_line += " | ";
schema_line += ps.cache.columns[i].first + " (" + ps.cache.columns[i].second + ")";
}
ImGui::TextDisabled("%s", schema_line.c_str());
ImGui::Separator();
}
// Data table via data_table::render.
if (!ps.cache.columns.empty()) {
int ncols = static_cast<int>(ps.cache.columns.size());
int nrows = static_cast<int>(ps.cache.rows.size());
data_table::TableInput tbl;
tbl.name = "preview";
tbl.headers.reserve(ncols);
tbl.types.reserve(ncols);
tbl.column_specs.resize(ncols);
for (int i = 0; i < ncols; i++) {
tbl.headers.push_back(ps.cache.columns[i].first);
tbl.types.push_back(data_table::ColumnType::String);
tbl.column_specs[i].id = ps.cache.columns[i].first;
}
tbl.rows = nrows;
tbl.cols = ncols;
// Build flat backing array row-major.
g_back_preview.clear();
g_back_preview.reserve(static_cast<size_t>(nrows) * ncols);
for (auto& row : ps.cache.rows) {
for (int c = 0; c < ncols; c++) {
g_back_preview.push_back(c < static_cast<int>(row.size()) ? row[c] : "");
}
}
cells_to_ptrs(g_back_preview, g_ptrs_preview);
tbl.cells = g_ptrs_preview.data();
data_table::render("##dt_preview", {tbl}, g_st_preview, nullptr, true);
}
// Pagination controls.
ImGui::Separator();
{
long long showing_from = ps.cache.offset + 1;
long long showing_to = ps.cache.offset + static_cast<long long>(ps.cache.rows.size());
long long total = ps.cache.total_rows;
if (ps.cache.rows.empty()) showing_from = 0;
char info[128];
std::snprintf(info, sizeof(info), "showing %lld-%lld of %lld",
showing_from, showing_to, total);
ImGui::TextDisabled("%s", info);
ImGui::SameLine();
bool can_prev = (ps.offset > 0);
bool can_next = (ps.offset + ps.limit < static_cast<int>(ps.cache.total_rows));
if (!can_prev) ImGui::BeginDisabled();
if (ImGui::SmallButton("< Prev")) {
ps.offset = std::max(0, ps.offset - ps.limit);
ps.loaded = false;
ps.error.clear();
ps.cache = data_factory::TablePreview{};
}
if (!can_prev) ImGui::EndDisabled();
ImGui::SameLine();
if (!can_next) ImGui::BeginDisabled();
if (ImGui::SmallButton("Next >")) {
ps.offset += ps.limit;
ps.loaded = false;
ps.error.clear();
ps.cache = data_factory::TablePreview{};
}
if (!can_next) ImGui::EndDisabled();
}
ImGui::End();
tabs.h
+5
View File
@@ -47,6 +47,9 @@ void draw_transformers(const std::string& api_url,
void draw_databases(const std::string& api_url,
const std::vector<data_factory::DatabaseInfo>& dbs);
void draw_tables(const std::string& api_url,
const std::vector<data_factory::TableEntry>& tables);
void draw_sinks(const std::string& api_url,
const std::vector<data_factory::Node>& nodes,
const std::vector<data_factory::Run>& runs_all);
@@ -59,4 +62,6 @@ void draw_node_detail_panel(const std::string& api_url,
const std::vector<data_factory::Run>& runs_all,
bool* p_open);
void draw_table_preview_panel(const std::string& api_url, bool* p_open);
} // namespace data_factory_ui