fix(index): prevent INDEX ON from corrupting source table indexes

src/abi/ace_exports.cpp

@@ -893,11 +893,29 @@ UNSIGNED32 sqlite_aggregate(
         return false;
     };
 
+    // Resolve a spec's field to its canonical, schema-validated column name.
+    // An unknown name must be rejected, never concatenated into the SQL: a
+    // double-quoted token that is not a real column is silently treated as a
+    // string literal by SQLite (TOTAL("NOPE") -> 0), and a quote in the name
+    // would break out of the identifier. Empty field is valid only for COUNT.
+    auto resolve_col = [&](const openads::engine::AggSpec& s,
+                           std::string& col) -> bool {
+        if (s.field.empty())
+            return s.fn == openads::engine::AggFn::Count;
+        for (const auto& f : st->fields) {
+            if (iequal(f.name, s.field)) { col = "\"" + f.name + "\""; return true; }
+        }
+        return false;
+    };
+
     std::string sql = "SELECT ";
     for (std::size_t i = 0; i < specs->size(); ++i) {
         if (i) sql += ", ";
         const auto& s   = (*specs)[i];
-        const std::string col = "\"" + s.field + "\"";
+        std::string col;
+        if (!resolve_col(s, col))
+            return fail(openads::AE_INTERNAL_ERROR,
+                        ("sqlite_aggregate: invalid field " + s.field).c_str());
         switch (s.fn) {
             case openads::engine::AggFn::Count:
                 sql += s.field.empty() ? "COUNT(*)" : ("COUNT(" + col + ")");
@@ -13537,6 +13555,18 @@ UNSIGNED32 AdsExecuteSQLDirect(ADSHANDLE hStatement, UNSIGNED8* pucSQL,
             if (static_cast<Connection*>(p) == c) conn_h = h;
         });
         if (conn_h == 0) return fail(openads::AE_INVALID_CONNECTION_HANDLE, "");
+
+        // M13.1 — validate table name is not a SELECT result.
+        // INDEX ON (SELECT ...) would open the source table file instead
+        // of using the materialized cursor → corrupts the source indexes.
+        // Enforce that table name is a simple identifier/filename.
+        const auto& tname = ci.value().table;
+        if (tname.empty() || tname[0] == '(' || tname.find("SELECT") != std::string::npos) {
+            return fail(openads::AE_SYNTAX_ERROR,
+                "INDEX ON requires a table name, not a SELECT result; "
+                "use SELECT ... ORDER BY/DISTINCT/LIMIT to materialize first");
+        }
+
         std::vector<UNSIGNED8> name_buf(ci.value().table.size() + 1, 0);
         std::memcpy(name_buf.data(), ci.value().table.data(),
                     ci.value().table.size());

src/network/session.cpp

@@ -1866,19 +1866,36 @@ DispatchResult Session::dispatch(const Frame& f) {
                 std::vector<std::int32_t> fidx;
                 accs.reserve(specs.size());
                 fidx.reserve(specs.size());
+                // Validate every spec before touching the table: an unknown
+                // field name (field_index -> -1) must be rejected, never folded
+                // as COUNT(*) -- that would silently return the row count for a
+                // typo'd or injected field. Only COUNT may omit the field.
+                bool specs_ok = true;
                 for (const auto& s : specs) {
-                    std::int32_t fi =
-                        s.field.empty() ? -1 : tbl->field_index(s.field);
-                    bool numeric = false;
-                    if (fi >= 0) {
-                        const auto& fd = tbl->field_descriptor(
-                            static_cast<std::uint16_t>(fi));
-                        numeric = field_is_numeric(fd.type);
+                    const auto fn = static_cast<openads::engine::AggFn>(s.fn);
+                    if (s.field.empty()) {
+                        if (fn != openads::engine::AggFn::Count) {
+                            reply = err("Aggregate: empty field only valid for "
+                                        "COUNT");
+                            specs_ok = false;
+                            break;
+                        }
+                        fidx.push_back(-1);
+                        accs.emplace_back(fn, false);
+                        continue;
+                    }
+                    std::int32_t fi = tbl->field_index(s.field);
+                    if (fi < 0) {
+                        reply = err("Aggregate: unknown field " + s.field);
+                        specs_ok = false;
+                        break;
                     }
-                    accs.emplace_back(
-                        static_cast<openads::engine::AggFn>(s.fn), numeric);
+                    const auto& fd =
+                        tbl->field_descriptor(static_cast<std::uint16_t>(fi));
+                    accs.emplace_back(fn, field_is_numeric(fd.type));
                     fidx.push_back(fi);
                 }
+                if (!specs_ok) break;
 
                 // Scan the whole table once, restoring the cursor afterwards.
                 std::uint32_t saved   = tbl->recno();

src/sql_backend/postgres_connection.cpp

@@ -428,14 +428,32 @@ PostgresConnection::aggregate(PostgresTable* tbl,
         return false;
     };
 
+    // Resolve a spec's field to its canonical, schema-validated column name and
+    // quote it. An unknown name must be rejected, never concatenated raw: a
+    // crafted field could otherwise inject SQL. Empty field is only for COUNT.
+    auto resolve_col = [&](const engine::AggSpec& s, std::string& col) -> bool {
+        if (s.field.empty()) return s.fn == engine::AggFn::Count;
+        for (const auto& f : tbl->fields) {
+            if (f.name.size() != s.field.size()) continue;
+            bool same = true;
+            for (std::size_t j = 0; j < s.field.size(); ++j)
+                if (std::toupper(static_cast<unsigned char>(f.name[j])) !=
+                    std::toupper(static_cast<unsigned char>(s.field[j]))) {
+                    same = false; break;
+                }
+            if (same) { col = quote_ident(f.name); return true; }
+        }
+        return false;
+    };
+
     std::string sql = "SELECT ";
     for (std::size_t i = 0; i < specs.size(); ++i) {
         if (i) sql += ", ";
         const auto& s = specs[i];
-        // Leave the column unquoted so PostgreSQL case-folds it (the ABI hands
-        // field names in upper case, e.g. "QTY", but the column is "qty");
-        // this matches how the translated WHERE refers to the same column.
-        const std::string col = s.field;
+        std::string col;
+        if (!resolve_col(s, col))
+            return util::Error{5001, 0,
+                               "invalid postgres aggregate field: " + s.field, ""};
         switch (s.fn) {
             case engine::AggFn::Count:
                 sql += s.field.empty() ? "COUNT(*)" : ("COUNT(" + col + ")");

tests/unit/abi_aggregate_postgres_test.cpp

@@ -133,6 +133,31 @@ TEST_CASE("Tier-3 push-down: PostgreSQL untranslatable FOR declines") {
     CHECK(hRes == 0);
 }
 
+// Regression for the Tier-3 review (#113): the field name must be validated
+// against the cached schema before it is concatenated into the SQL, otherwise
+// a typo'd or injection-shaped field reaches PostgreSQL raw.
+TEST_CASE("Tier-3 push-down: PostgreSQL rejects an unknown aggregate field") {
+    PgFixture fx;
+    fx.open();
+
+    UNSIGNED8 forc[] = "";
+    UNSIGNED8 spec[] = "SUM:NOPE";
+    ADSHANDLE hRes   = 0;
+    CHECK(AdsAggregate(fx.hTable, forc, spec, &hRes) != AE_SUCCESS);
+    CHECK(hRes == 0);
+}
+
+TEST_CASE("Tier-3 push-down: PostgreSQL rejects empty field for non-COUNT") {
+    PgFixture fx;
+    fx.open();
+
+    UNSIGNED8 forc[] = "";
+    UNSIGNED8 spec[] = "SUM:";
+    ADSHANDLE hRes   = 0;
+    CHECK(AdsAggregate(fx.hTable, forc, spec, &hRes) != AE_SUCCESS);
+    CHECK(hRes == 0);
+}
+
 #else
 
 TEST_CASE("Tier-3 aggregate push-down: postgresql backend disabled at compile time") {

tests/unit/abi_aggregate_sqlite_test.cpp

@@ -10,6 +10,7 @@
 #include <cstring>
 #include <filesystem>
 #include <string>
+#include <vector>
 
 namespace fs = std::filesystem;
 
@@ -121,6 +122,46 @@ TEST_CASE("Tier-3 push-down: SQLite zero matches -> 0 / 0 / empty") {
     REQUIRE(AdsAggregateClose(hRes) == AE_SUCCESS);
 }
 
+// Regression for the Tier-3 review (#113): the SQLite push-down concatenated
+// the spec's field name straight into `TOTAL("<field>")` etc. A field that is
+// not a real column (a typo, or an injection probe carrying a quote) must be
+// rejected by validating against the cached schema BEFORE any SQL is built --
+// never reach the backend as a raw identifier.
+TEST_CASE("Tier-3 push-down: SQLite rejects an unknown aggregate field") {
+    SqliteFixture fx;
+    fx.open();
+
+    UNSIGNED8 forc[] = "";
+    UNSIGNED8 spec[] = "SUM:NOPE";       // NOPE is not a column of items
+    ADSHANDLE hRes   = 0;
+    CHECK(AdsAggregate(fx.hTable, forc, spec, &hRes) != AE_SUCCESS);
+    CHECK(hRes == 0);
+}
+
+TEST_CASE("Tier-3 push-down: SQLite rejects an injection-shaped field name") {
+    SqliteFixture fx;
+    fx.open();
+
+    UNSIGNED8 forc[] = "";
+    // A quote in the field name would break out of the quoted identifier if it
+    // ever reached SQL. Validation must reject it up front.
+    UNSIGNED8 spec[] = "MIN:nm\")--";
+    ADSHANDLE hRes   = 0;
+    CHECK(AdsAggregate(fx.hTable, forc, spec, &hRes) != AE_SUCCESS);
+    CHECK(hRes == 0);
+}
+
+TEST_CASE("Tier-3 push-down: SQLite rejects empty field for non-COUNT") {
+    SqliteFixture fx;
+    fx.open();
+
+    UNSIGNED8 forc[] = "";
+    UNSIGNED8 spec[] = "SUM:";           // empty field, non-COUNT
+    ADSHANDLE hRes   = 0;
+    CHECK(AdsAggregate(fx.hTable, forc, spec, &hRes) != AE_SUCCESS);
+    CHECK(hRes == 0);
+}
+
 TEST_CASE("Tier-3 push-down: untranslatable FOR declines (caller falls back)") {
     SqliteFixture fx;
     fx.open();

tests/unit/abi_aggregate_test.cpp

@@ -171,6 +171,45 @@ TEST_CASE("AdsAggregate remote wire: COUNT/SUM/AVG/MIN/MAX over a FOR predicate"
     CHECK(AdsAggregateCount(hRes, &dummy) != AE_SUCCESS);   // handle freed
 }
 
+// ── 4. Remote wire: an unknown field must be rejected, not silently counted ──
+// Regression for the Tier-3 review (#113): the server resolved each spec's
+// field with field_index(), which returns -1 for a name that isn't in the
+// table -- and the scan loop treated fidx < 0 as COUNT(*). So `SUM:NOPE`
+// silently returned the *row count* as a numeric total with AE_SUCCESS,
+// masking a typo'd or injected field name. It must fail instead.
+TEST_CASE("AdsAggregate remote wire: unknown field is rejected, not silent COUNT") {
+    agg_wipe();
+    auto dir = agg_tmp_dir();
+    seed_agg_fixture(dir);
+
+    RemoteFixture fx;
+    fx.open(dir);
+
+    UNSIGNED8 forc[] = "";
+    UNSIGNED8 spec[] = "SUM:NOPE";        // NOPE is not a column of agg.dbf
+    ADSHANDLE hRes   = 0;
+    CHECK(AdsAggregate(fx.hTable, forc, spec, &hRes) != AE_SUCCESS);
+    CHECK(hRes == 0);
+}
+
+// ── 5. Remote wire: an empty field is only valid for COUNT ───────────────────
+// `SUM:` / `AVG:` / `MIN:` / `MAX:` with no field used to fall through the same
+// fidx < 0 path and be folded as COUNT(*). Only COUNT may omit the field.
+TEST_CASE("AdsAggregate remote wire: empty field for non-COUNT is rejected") {
+    agg_wipe();
+    auto dir = agg_tmp_dir();
+    seed_agg_fixture(dir);
+
+    RemoteFixture fx;
+    fx.open(dir);
+
+    UNSIGNED8 forc[] = "";
+    UNSIGNED8 spec[] = "SUM:";            // empty field, non-COUNT
+    ADSHANDLE hRes   = 0;
+    CHECK(AdsAggregate(fx.hTable, forc, spec, &hRes) != AE_SUCCESS);
+    CHECK(hRes == 0);
+}
+
 // ── 3. Remote wire: zero matches ─────────────────────────────────────────────
 TEST_CASE("AdsAggregate remote wire: zero matches -> 0 / 0 / empty") {
     agg_wipe();