feat(ingest)!: destination flags follow the family

The service now takes `table` from the families that name the one table they
write, and an optional `table_prefix` from the ones whose tables are named
after the source. Sending the wrong one is refused with the field named, so
the flags follow: `--dest-table` for the first, `--dest-table-prefix` for the
second, each saying in help which families it applies to.

Omitting the prefix keeps the raw source names, which is what someone syncing
raw tables wants and what happens today. Listings present a prefix as a
prefix, so nobody reads it as a table that will exist.

Also: `ingest create --help` claimed a one-time ingest "runs immediately and
reports its initial_run_id". It is always null -- the scheduler dispatches --
and the help now says what actually happens. And `datasource fields` dropped
the field descriptions the service sends from its table view, which is where
the useful text is: `root_uri`'s "object keys are recorded relative to it, so
moving it starts a fresh read" was visible only under -o json.

Author: anoop-narang

README.md

@@ -115,7 +115,13 @@ hotdata datasource list                     # ids, families, states
 # 3. Ingest — what to load, and where it lands. Once:
 hotdata ingest create --source "prod postgres" --table orders --database-id db_123
 
-#    …or on a schedule, kept fresh:
+#    Source tables keep their own names. Put several under a common prefix —
+#    orders lands as raw_orders, customers as raw_customers:
+hotdata ingest create --source "prod postgres" --database-id db_123 \
+  --table orders --table customers --dest-table-prefix raw
+
+#    A source that lands ONE table is told that table's name instead, and can
+#    be kept fresh on a schedule:
 hotdata ingest create --source ds_01J --type continuous --database-id db_123 \
   --format parquet --glob "orders/**/*.parquet" --dest-table orders_raw --every 5m
 

src/client/ingest.rs

@@ -65,10 +65,22 @@ impl IngestError {
             // or a follow-up hint should branch on — so print both.
             IngestError::Http { status, body } => {
                 let message = util::api_error(body.clone());
-                match util::error_code(body) {
+                let mut rendered = match util::error_code(body) {
                     Some(code) => format!("HTTP {status}: {message} ({code})"),
                     None => format!("HTTP {status}: {message}"),
+                };
+                // The message says which payload was rejected; the details say
+                // which field, and only together are they an instruction. One
+                // line each, indented under it, because a rejected body can
+                // name several fields and a joined sentence hides all but the
+                // first.
+                for (field, msg) in util::error_fields(body) {
+                    rendered.push_str(&match field.as_str() {
+                        "" => format!("\n  {msg}"),
+                        f => format!("\n  {f}: {msg}"),
+                    });
                 }
+                rendered
             }
             IngestError::Connection(e) => format!("connection error: {e}"),
             IngestError::Decode(e) => format!("malformed response: {e}"),
@@ -153,6 +165,20 @@ impl IngestError {
                         .dark_grey()
                     );
                 }
+                // The destination is per-family, and the pair of flags that
+                // build it is the thing to re-read: a family lands one table
+                // or a table per source table, never both, so a rejected
+                // `table`/`table_prefix` is the other flag's job.
+                Some("invalid_destination") => {
+                    eprintln!(
+                        "{}",
+                        "--dest-table names ONE table (buckets, Delta, a --raw-sql result); \
+                         --dest-table-prefix prefixes the tables a source that lands several \
+                         produces (SQL --table/--sql, Iceberg, DuckLake, Kafka, REST). The \
+                         family's own field reference: hotdata datasource fields <family>."
+                            .dark_grey()
+                    );
+                }
                 Some("immutable_ingest_definition") => {
                     eprintln!(
                         "{}",
@@ -1002,16 +1028,24 @@ pub struct FamiliesResponse {
 /// `format: password` sits BESIDE the `anyOf`, not inside one of its members —
 /// which is what makes an optional secret's marker easy to miss, and a secret
 /// whose marker is missed is one the guided flow echoes to the terminal.
+///
+/// The `description` on a property is the service's own sentence about the
+/// field, written on the model that validates it. Some properties have one and
+/// some do not, which is the pair of cases the reference has to render — a
+/// fixture where none did is how the field reference came to drop all of them
+/// from its table and keep them only under `-o json`.
 #[cfg(test)]
 pub const FAMILY_REFERENCE_BODY: &str = r#"{
   "family": "sql",
   "config_schema": {
     "additionalProperties": false,
     "properties": {
       "dialect": {"enum": ["postgres", "mysql", "duckdb"], "title": "Dialect",
-                  "type": "string"},
+                  "type": "string",
+                  "description": "Which SQL engine this is. The rest of `config` is read in its terms."},
       "host": {"anyOf": [{"type": "string"}, {"type": "null"}], "default": null,
-               "title": "Host"},
+               "title": "Host",
+               "description": "Server hostname. Omit for engines addressed by account or path."},
       "port": {"anyOf": [{"maximum": 65535, "minimum": 1, "type": "integer"},
                          {"type": "null"}], "default": null, "title": "Port"},
       "options": {"additionalProperties": true, "title": "Options", "type": "object"}
@@ -1040,7 +1074,8 @@ pub const FAMILY_REFERENCE_BODY: &str = r#"{
          "schema": {"anyOf": [{"type": "string"}, {"type": "null"}],
                     "default": null, "title": "Schema"},
          "tables": {"items": {"type": "string"}, "minItems": 1,
-                    "title": "Tables", "type": "array"},
+                    "title": "Tables", "type": "array",
+                    "description": "Source tables to read. Each lands in a destination table of the same name."},
          "limit": {"anyOf": [{"minimum": 1, "type": "integer"}, {"type": "null"}],
                    "default": null, "title": "Limit"}
        },
@@ -1760,6 +1795,37 @@ mod tests {
         }
     }
 
+    /// A 422 about ONE field must say which field.
+    ///
+    /// The message half names the payload and the family and stops there — by
+    /// design, since `details` is where the field paths are. Printing only the
+    /// message left a caller who sent a destination field the family does not
+    /// have with "invalid destination for family 'iceberg'" and nothing to act
+    /// on, which is the same amount of information as no error text at all.
+    #[test]
+    fn a_field_level_rejection_names_the_field_it_rejected() {
+        let err = IngestError::Http {
+            status: 422,
+            body: r#"{"error": {"code": "invalid_destination",
+                      "message": "invalid destination for family 'iceberg'",
+                      "details": {"errors": [
+                        {"field": "table", "message": "Extra inputs are not permitted"}]}}}"#
+                .into(),
+        };
+        let rendered = err.message();
+        assert!(rendered.contains("invalid_destination"), "{rendered}");
+        assert!(
+            rendered.contains("\n  table: Extra inputs are not permitted"),
+            "{rendered}"
+        );
+        // An envelope with no field errors is unchanged — one line, as before.
+        let plain = IngestError::Http {
+            status: 404,
+            body: r#"{"error":{"code":"run_not_found","message":"no run 'nope'"}}"#.into(),
+        };
+        assert_eq!(plain.message(), "HTTP 404: no run 'nope' (run_not_found)");
+    }
+
     // --- catalog -----------------------------------------------------------
 
     #[test]

src/commands/datasource.rs

@@ -920,6 +920,16 @@ fn print_family_reference(r: &FamilyReference) {
     print_schema(&r.credentials_schema);
     section("SELECTOR", "hotdata ingest create --selector");
     print_schema(&r.selector_schema);
+    // The destination is per-family too — one family names a table, another
+    // takes at most a prefix for the tables it lands. Read out of the
+    // passthrough map rather than a field of our own: this section exists
+    // BECAUSE the service is the only thing that can be right about which,
+    // and a family whose reference does not describe it gets no section
+    // rather than a CLI-side answer.
+    if let Some(schema) = r.extra.get("destination_schema") {
+        section("DESTINATION", "hotdata ingest create --destination");
+        print_schema(schema);
+    }
     println!();
     hint(&format!(
         "'hotdata datasource fields {} -o json' prints the JSON Schema itself, \
@@ -975,7 +985,10 @@ fn print_schema(schema: &serde_json::Value) {
             hint("  (no fields)");
             continue;
         }
-        crate::output::table::print(&["FIELD", "TYPE", "REQUIRED", "DEFAULT"], &rows);
+        crate::output::table::print(
+            &["FIELD", "TYPE", "REQUIRED", "DEFAULT", "DESCRIPTION"],
+            &rows,
+        );
     }
 }
 
@@ -1034,11 +1047,33 @@ fn field_rows(schema: &serde_json::Value) -> Vec<Vec<String>> {
                 type_label(prop),
                 yes_no(required.iter().any(|r| r == name)),
                 default_cell(prop),
+                description_cell(prop),
             ]
         })
         .collect()
 }
 
+/// What the field is FOR, as the service describes it.
+///
+/// Whole, not summarised: these sentences are written on the models that
+/// validate the request and they carry the part no type or default can — which
+/// engines want a host, what a root uri does to object keys, what a prefix is
+/// relative to. Trimming them to a column width would drop the second sentence,
+/// which is where that usually lives. The table wraps this cell instead, and a
+/// terminal too narrow to hold it says so and points at `-o json`.
+///
+/// Newlines are collapsed because a description written across source lines is
+/// one sentence, and a cell that kept the breaks would set its own row height
+/// rather than wrap to the width it was given.
+fn description_cell(prop: &serde_json::Value) -> String {
+    match prop.get("description").and_then(|d| d.as_str()) {
+        Some(text) if !text.trim().is_empty() => {
+            text.split_whitespace().collect::<Vec<_>>().join(" ")
+        }
+        _ => "-".into(),
+    }
+}
+
 fn required_names(schema: &serde_json::Value) -> Vec<String> {
     schema
         .get("required")
@@ -1461,21 +1496,60 @@ mod tests {
     }
 
     #[test]
-    fn field_rows_carry_the_type_the_requirement_and_the_default() {
+    fn field_rows_carry_the_type_the_requirement_the_default_and_the_description() {
         let reference = sql_reference();
         let rows = field_rows(&reference.config_schema);
 
         // An enum's members ARE the type: "string" would leave the caller
-        // guessing at the one thing a 422 will be about.
+        // guessing at the one thing a 422 will be about. The description is
+        // the service's own sentence about the field, and it is the half of
+        // the row that says what the field is FOR — it used to reach `-o json`
+        // and nothing else.
         assert_eq!(
             row(&rows, "dialect"),
-            ["dialect", "postgres | mysql | duckdb", "yes", "-"]
+            [
+                "dialect",
+                "postgres | mysql | duckdb",
+                "yes",
+                "-",
+                "Which SQL engine this is. The rest of `config` is read in its terms."
+            ]
         );
         // Optional fields arrive as a union against null. The null member is
         // the REQUIRED column's job and must not reach the type.
-        assert_eq!(row(&rows, "host"), ["host", "string", "no", "-"]);
-        assert_eq!(row(&rows, "port"), ["port", "integer", "no", "-"]);
-        assert_eq!(row(&rows, "options"), ["options", "object", "no", "-"]);
+        assert_eq!(
+            row(&rows, "host"),
+            [
+                "host",
+                "string",
+                "no",
+                "-",
+                "Server hostname. Omit for engines addressed by account or path."
+            ]
+        );
+        // A field the service describes with nothing gets a dash, like every
+        // other empty cell in the CLI — never a blank column.
+        assert_eq!(row(&rows, "port"), ["port", "integer", "no", "-", "-"]);
+        assert_eq!(row(&rows, "options"), ["options", "object", "no", "-", "-"]);
+    }
+
+    #[test]
+    fn a_description_written_across_lines_is_one_cell() {
+        // Pydantic descriptions are written as adjacent source strings and
+        // arrive with the breaks in them. A cell keeping those breaks sets its
+        // own row height instead of wrapping to the width the table gave it.
+        let prop = serde_json::json!({
+            "type": "string",
+            "description": "Full DSN, e.g. postgresql://user@host/db.\nReplaces the\n  fields above."
+        });
+        assert_eq!(
+            description_cell(&prop),
+            "Full DSN, e.g. postgresql://user@host/db. Replaces the fields above."
+        );
+        assert_eq!(
+            description_cell(&serde_json::json!({"type": "string"})),
+            "-"
+        );
     }
 
     #[test]
@@ -1489,17 +1563,60 @@ mod tests {
         let (tables_label, tables) = &variants[0];
         assert_eq!(tables_label.as_deref(), Some("mode = tables"));
         let rows = field_rows(tables);
-        assert_eq!(row(&rows, "tables"), ["tables", "string[]", "yes", "-"]);
-        assert_eq!(row(&rows, "mode"), ["mode", "tables", "no", "tables"]);
-        assert_eq!(row(&rows, "schema"), ["schema", "string", "no", "-"]);
+        assert_eq!(
+            row(&rows, "tables"),
+            [
+                "tables",
+                "string[]",
+                "yes",
+                "-",
+                "Source tables to read. Each lands in a destination table of the same name."
+            ]
+        );
+        assert_eq!(row(&rows, "mode"), ["mode", "tables", "no", "tables", "-"]);
+        assert_eq!(row(&rows, "schema"), ["schema", "string", "no", "-", "-"]);
 
         // Each form has its own required set: `sql` is required in the query
         // form and absent from the other, which merging them would hide.
         let (query_label, query) = &variants[1];
         assert_eq!(query_label.as_deref(), Some("mode = query"));
         let rows = field_rows(query);
-        assert_eq!(row(&rows, "sql"), ["sql", "string", "yes", "-"]);
-        assert_eq!(row(&rows, "mode"), ["mode", "query", "yes", "-"]);
+        assert_eq!(row(&rows, "sql"), ["sql", "string", "yes", "-", "-"]);
+        assert_eq!(row(&rows, "mode"), ["mode", "query", "yes", "-", "-"]);
+    }
+
+    /// The destination is per-family, and a family that describes one is
+    /// carried through the passthrough map rather than a field of the CLI's:
+    /// which of `table` / `table_prefix` a family takes is the service's
+    /// answer, and a CLI-side copy of it is the thing this command replaces.
+    #[test]
+    fn a_family_that_describes_its_destination_is_carried_through() {
+        let reference = sql_reference();
+        assert!(
+            reference.extra.get("destination_schema").is_none(),
+            "the pinned body describes no destination; the section must not invent one"
+        );
+
+        let described: FamilyReference = serde_json::from_str(
+            r#"{"family": "iceberg", "destination_schema": {
+                 "properties": {"table_prefix": {"type": "string"}},
+                 "additionalProperties": false, "type": "object"}}"#,
+        )
+        .unwrap();
+        let schema = described
+            .extra
+            .get("destination_schema")
+            .expect("carried through untyped");
+        assert_eq!(
+            field_rows(schema),
+            vec![vec![
+                "table_prefix".to_string(),
+                "string".into(),
+                "no".into(),
+                "-".into(),
+                "-".into()
+            ]]
+        );
     }
 
     #[test]