Skip to content

ECOPROJECT-4839 | feat: add collection_id to the duck_db ingestion logic #1304

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
amalimov wants to merge 1 commit into
base: main
Choose a base branch
from
Open

ECOPROJECT-4839 | feat: add collection_id to the duck_db ingestion logic #1304

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 15 additions & 2 deletions pkg/duckdb_parser/builder.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,8 @@ func NewBuilder() *QueryBuilder {
}

type ingestParams struct {
FilePath string
FilePath string
CollectionID int64
}

// CreateSchemaQuery returns queries to create all RVTools tables with proper schema.
Expand All @@ -55,8 +56,20 @@ func (b *QueryBuilder) IngestRvtoolsQuery(filePath string) (string, error) {
}

// IngestSqliteQuery returns a query that creates RVTools-shaped tables from a forklift SQLite database.
// Use IngestSqliteQueryWithCollection to hash VM IDs and stamp collection_id.
func (b *QueryBuilder) IngestSqliteQuery(filePath string) (string, error) {
return b.buildQuery("ingest_sqlite", mustGetTemplate("ingest_sqlite"), ingestParams{FilePath: filePath})
return b.IngestSqliteQueryWithCollection(filePath, 0)
}

// IngestSqliteQueryWithCollection returns the same query as IngestSqliteQuery but, when
// collectionID > 0, hashes each VM's "VM ID" to md5('{collectionID}_{vSphere_MOID}'),
// stores the original MOID in the vmmoid column, and writes collection_id.
func (b *QueryBuilder) IngestSqliteQueryWithCollection(filePath string, collectionID int64) (string, error) {
if collectionID < 0 {
return "", fmt.Errorf("collectionID must be non-negative, got %d", collectionID)
}
return b.buildQuery("ingest_sqlite", mustGetTemplate("ingest_sqlite"),
ingestParams{FilePath: filePath, CollectionID: collectionID})
}

// queryParams holds all template parameters for queries.
Expand Down
29 changes: 22 additions & 7 deletions pkg/duckdb_parser/ingest.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,12 @@ func isCriticalStatement(stmt string) bool {
return true
}
}
// INSERT INTO vinfo is critical only when it carries collection_id/vmmoid columns
// (SQLite collection path). The RVTools INSERT INTO vinfo does not include these columns
// and may legitimately fail when the source Excel is malformed.
if strings.Contains(upperStmt, "INSERT INTO VINFO") && strings.Contains(upperStmt, "VMMOID") {
return true
}
Comment on lines +30 to +35

@coderabbitai coderabbitai Bot Jul 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift

Do not use VMMOID as the proxy for RVTools-only leniency.

This also makes the SQLite collectionID=0 INSERT INTO vinfo non-critical, because that template path has no vmmoid column. A failed SQLite vinfo load can now continue without an ingestion error, which is not the same as preserving existing IngestSqlite behavior. Scope the relaxed behavior by ingestion source/policy instead of statement contents. As per path instructions, errors must be checked, wrapped with context, and propagated.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/duckdb_parser/ingest.go` around lines 30 - 35, The criticality check in
the ingest flow is too broad because it keys off `VMMOID` inside the statement,
which wrongly relaxes SQLite `INSERT INTO vinfo` failures. Update the logic
around the ingestion path in `duckdb_parser/ingest.go` so the relaxed handling
is driven by the ingestion source/policy (for example, the RVTools path) rather
than SQL text contents. Keep `IngestSqlite` behavior unchanged: check every
error, wrap it with context, and propagate it unless the source-specific policy
explicitly marks it non-critical.

Source: Path instructions

return false
}

Expand Down Expand Up @@ -81,23 +87,32 @@ func (p *Parser) IngestRvTools(ctx context.Context, excelFile string) (Validatio
return result, nil
}

// IngestSqlite ingests data from a forklift SQLite database, runs VM validation if a validator
// is configured, and validates the schema for required tables/columns.
// Returns a ValidationResult with errors (fatal) and warnings (non-fatal).
// If ValidationResult.HasErrors() is true, the inventory cannot be built.
// IngestSqlite ingests data from a forklift SQLite database with no collection context.
// Existing behaviour is preserved exactly. Use IngestSqliteWithCollection when the
// agent has an active collection and wants hashed VM IDs.
func (p *Parser) IngestSqlite(ctx context.Context, sqliteFile string) (ValidationResult, error) {
query, err := p.builder.IngestSqliteQuery(sqliteFile)
return p.IngestSqliteWithCollection(ctx, sqliteFile, 0)
}

// IngestSqliteWithCollection ingests from a forklift SQLite database.
// When collectionID > 0 every "VM ID" in vinfo and relational tables is replaced
// with md5('{collectionID}_{original_moid}'); the original MOID is written to
// the vmmoid column and collection_id is set. When collectionID = 0 the
// behaviour is identical to IngestSqlite (no hashing, no extra columns written).
func (p *Parser) IngestSqliteWithCollection(ctx context.Context, sqliteFile string, collectionID int64) (ValidationResult, error) {
if collectionID < 0 {
return ValidationResult{}, fmt.Errorf("collectionID must be non-negative, got %d", collectionID)
}
query, err := p.builder.IngestSqliteQueryWithCollection(sqliteFile, collectionID)
if err != nil {
return ValidationResult{}, fmt.Errorf("building sqlite ingestion query: %w", err)
}
if err := p.executeStatements(ctx, query); err != nil {
return ValidationResult{}, fmt.Errorf("ingesting sqlite data: %w", err)
}

// Validate schema against vinfo (SQLite inserts directly into vinfo, no vinfo_raw)
result := p.ValidateSchema(ctx, "vinfo")

// Only run post-ingestion steps if schema is valid (we have VMs to process)
if result.IsValid() {
if err := p.populateComplexity(ctx); err != nil {
return result, fmt.Errorf("populating complexity: %w", err)
Expand Down
132 changes: 132 additions & 0 deletions pkg/duckdb_parser/ingest_sqlite_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package duckdb_parser

import (
"context"
"crypto/md5"
"database/sql"
"fmt"
"path/filepath"
Expand Down Expand Up @@ -389,3 +390,134 @@ func TestIngestSqlite_PopulateVCluster_SkipsWhenAlreadyPopulated(t *testing.T) {
require.NoError(t, db.QueryRowContext(ctx, `SELECT "Object ID" FROM vcluster WHERE "Name" = 'existing-cluster'`).Scan(&objectID))
assert.Equal(t, "sentinel-id", objectID, "pre-existing row must not be overwritten")
}

// addCollectionColumns adds vmmoid and collection_id columns to vinfo so that
// IngestSqliteWithCollection can write to them. In production these columns are
// added by agent migration 026; in tests we add them manually.
func addCollectionColumns(t *testing.T, db *sql.DB) {
t.Helper()
ctx := context.Background()
_, err := db.ExecContext(ctx, `ALTER TABLE vinfo ADD COLUMN IF NOT EXISTS vmmoid VARCHAR`)
require.NoError(t, err, "adding vmmoid column to vinfo")
_, err = db.ExecContext(ctx, `ALTER TABLE vinfo ADD COLUMN IF NOT EXISTS collection_id BIGINT`)
require.NoError(t, err, "adding collection_id column to vinfo")
}

func TestIngestSqliteWithCollection_HashesVMIDsAndSetsVmmoid(t *testing.T) {
const collectionID = int64(7)
ctx := context.Background()

parser, db, cleanup := setupTestParser(t, &testValidator{})
defer cleanup()

addCollectionColumns(t, db)

clusters := []sqliteCluster{
{id: "domain-c1", name: "cluster1", datacenter: "dc1"},
}
vms := []sqliteVM{
{id: "vm-001", name: "vm-1", clusterName: "cluster1"},
{id: "vm-002", name: "vm-2", clusterName: "cluster1"},
}
sqlitePath := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)

result, err := parser.IngestSqliteWithCollection(ctx, sqlitePath, collectionID)
require.NoError(t, err)
require.True(t, result.IsValid())

rows, err := db.QueryContext(ctx, `SELECT "VM ID", vmmoid, collection_id FROM vinfo`)
require.NoError(t, err)
defer func() { _ = rows.Close() }()

rowCount := 0
for rows.Next() {
rowCount++
var vmID, vmmoid string
var colID int64
require.NoError(t, rows.Scan(&vmID, &vmmoid, &colID))

// "VM ID" must be a 32-char lowercase hex md5 string.
assert.Len(t, vmID, 32, "VM ID should be a 32-char md5 hex string")
assert.Regexp(t, `^[0-9a-f]{32}$`, vmID, "VM ID should be a lowercase hex string")

// vmmoid is the original MOID (non-empty, not itself a 32-char hash).
assert.NotEmpty(t, vmmoid, "vmmoid should be the original MOID")
assert.NotEqual(t, 32, len(vmmoid), "vmmoid should not look like an md5 hash (original MOIDs are shorter)")

// collection_id must be set correctly.
assert.Equal(t, collectionID, colID, "collection_id should match the supplied collectionID")

// Hash must be deterministic: md5("{collectionID}_{vmmoid}") == vmID
expectedHash := fmt.Sprintf("%x", md5.Sum(fmt.Appendf(nil, "%d_%s", collectionID, vmmoid)))
assert.Equal(t, expectedHash, vmID, "VM ID should equal md5('%d_%s', collectionID, vmmoid)")
Comment on lines +450 to +452

@coderabbitai coderabbitai Bot Jul 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win

Assertion failure message won't interpolate values.

assert.Equal(t, expectedHash, vmID, "VM ID should equal md5('%d_%s', collectionID, vmmoid)") passes a single literal string as msgAndArgs; testify only applies fmt.Sprintf formatting when additional args follow the format string. On failure this message will print the literal text %d_%s instead of the actual collectionID/vmmoid values, making debugging harder.

🐛 Proposed fix 
-		assert.Equal(t, expectedHash, vmID, "VM ID should equal md5('%d_%s', collectionID, vmmoid)")
+		assert.Equal(t, expectedHash, vmID, "VM ID should equal md5(%d_%s)", collectionID, vmmoid)

As per path instructions, "Assertions must be specific and provide informative messages on failure."

📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
// Hash must be deterministic: md5("{collectionID}_{vmmoid}") == vmID
expectedHash := fmt.Sprintf("%x", md5.Sum(fmt.Appendf(nil, "%d_%s", collectionID, vmmoid)))
assert.Equal(t, expectedHash, vmID, "VM ID should equal md5('%d_%s', collectionID, vmmoid)")
// Hash must be deterministic: md5("{collectionID}_{vmmoid}") == vmID
expectedHash := fmt.Sprintf("%x", md5.Sum(fmt.Appendf(nil, "%d_%s", collectionID, vmmoid)))
assert.Equal(t, expectedHash, vmID, "VM ID should equal md5(%d_%s)", collectionID, vmmoid)
🧰 Tools
🪛 ast-grep (0.44.0)

[warning] 450-450: Detected use of the 'crypto/md5' package (md5.New / md5.Sum). MD5 is a cryptographically broken, collision-prone hash and must not be used for security purposes such as integrity checks, signatures, or password hashing. Use a strong hash from 'crypto/sha256' (sha256.New / sha256.Sum256) or 'crypto/sha512' instead; for passwords use a dedicated KDF such as golang.org/x/crypto/bcrypt or argon2.
Context: md5.Sum
Note: [CWE-327] Use of a Broken or Risky Cryptographic Algorithm.

(weak-hash-md5-go)

🪛 OpenGrep (1.23.0)

[WARNING] 451-451: MD5 and SHA1 are cryptographically broken and should not be used for security purposes. Use SHA-256 or stronger.

(coderabbit.crypto.go-weak-hash)

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/duckdb_parser/ingest_sqlite_test.go` around lines 450 - 452, The
assertion in ingest_sqlite_test.go uses assert.Equal with a literal message
string, so the failure text won’t interpolate collectionID and vmmoid. Update
the VM ID check in the test to pass a formatted message with the actual values
(or otherwise build the message before calling assert.Equal) so failures from
the deterministic hash check clearly show the collectionID and vmmoid used by
the test.

Source: Path instructions

}
require.NoError(t, rows.Err())
assert.Equal(t, len(vms), rowCount, "vinfo should have one row per non-template VM")
}

func TestIngestSqliteWithCollection_RelationalTablesMatchVinfoIDs(t *testing.T) {
const collectionID = int64(7)
ctx := context.Background()

parser, db, cleanup := setupTestParser(t, &testValidator{})
defer cleanup()

addCollectionColumns(t, db)

clusters := []sqliteCluster{
{id: "domain-c1", name: "cluster1", datacenter: "dc1"},
}
vms := []sqliteVM{
{id: "vm-001", name: "vm-1", clusterName: "cluster1"},
{id: "vm-002", name: "vm-2", clusterName: "cluster1"},
}
sqlitePath := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)

_, err := parser.IngestSqliteWithCollection(ctx, sqlitePath, collectionID)
require.NoError(t, err)

// Every vcpu "VM ID" must exist in vinfo "VM ID" — no orphaned rows.
var orphans int
err = db.QueryRowContext(ctx, `
SELECT COUNT(*) FROM vcpu
WHERE "VM ID" NOT IN (SELECT "VM ID" FROM vinfo)
`).Scan(&orphans)
require.NoError(t, err)
assert.Equal(t, 0, orphans, "all vcpu rows should reference a valid vinfo VM ID")
Comment on lines +458 to +486

@coderabbitai coderabbitai Bot Jul 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Exercise every collection-aware relationship, not only vcpu.

The template changed vcpu, vmemory, vdisk, and vnetwork, but this test only checks vcpu, and the fixture has empty disks/NICs. Add disk/NIC fixture data and assert no orphans for vmemory, vdisk, and vnetwork too. As per path instructions, tests should “Strive for high test coverage on critical logic paths.”

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/duckdb_parser/ingest_sqlite_test.go` around lines 458 - 486, The
collection-aware relationship test in
TestIngestSqliteWithCollection_RelationalTablesMatchVinfoIDs only validates
vcpu, so it misses the other affected tables. Extend the fixture data in
createTestSQLite for this test to include disk and NIC records, then add orphan
checks against vinfo for vmemory, vdisk, and vnetwork in the same test so every
collection-aware relationship is covered.

Source: Path instructions

}

func TestIngestSqliteWithCollection_ZeroCollectionIDMatchesIngestSqlite(t *testing.T) {
ctx := context.Background()

clusters := []sqliteCluster{
{id: "domain-c1", name: "cluster1", datacenter: "dc1"},
}
vms := []sqliteVM{
{id: "vm-001", name: "vm-1", clusterName: "cluster1"},
{id: "vm-002", name: "vm-2", clusterName: "cluster1"},
}

// Parser A uses IngestSqlite (the zero-collection-ID path).
parserA, dbA, cleanupA := setupTestParser(t, &testValidator{})
defer cleanupA()

// Parser B uses IngestSqliteWithCollection with collectionID=0.
parserB, dbB, cleanupB := setupTestParser(t, &testValidator{})
defer cleanupB()

sqlitePathA := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)
sqlitePathB := createTestSQLite(t, "vcenter-uuid-001", clusters, vms)

resultA, err := parserA.IngestSqlite(ctx, sqlitePathA)
require.NoError(t, err)

resultB, err := parserB.IngestSqliteWithCollection(ctx, sqlitePathB, 0)
require.NoError(t, err)

assert.Equal(t, resultA.IsValid(), resultB.IsValid(), "validity should match between IngestSqlite and IngestSqliteWithCollection(0)")

var countA, countB int
require.NoError(t, dbA.QueryRowContext(ctx, `SELECT COUNT(*) FROM vinfo`).Scan(&countA))
require.NoError(t, dbB.QueryRowContext(ctx, `SELECT COUNT(*) FROM vinfo`).Scan(&countB))
assert.Equal(t, countA, countB, "VM count should match between IngestSqlite and IngestSqliteWithCollection(0)")
Comment on lines +517 to +522

@coderabbitai coderabbitai Bot Jul 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Compare representative IDs, not just row counts.

Backward compatibility for collectionID=0 would still pass here if VM IDs or related rows changed but counts stayed the same. Compare ordered "VM ID" values, and at least one related-table "VM ID", between both parsers. As per path instructions, “Assertions must be specific and provide informative messages on failure.”

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/duckdb_parser/ingest_sqlite_test.go` around lines 517 - 522, The existing
comparison in the IngestSqlite test only checks validity and row counts, which
can miss regressions where IDs change but totals stay the same. Update the
assertions in the test around resultA/resultB and the vinfo query to compare
ordered representative "VM ID" values from both parsers, and also verify at
least one related-table "VM ID" matches between them; use specific, informative
assertion messages that clearly identify which ID set differed.

Source: Path instructions

}
Comment on lines +406 to +523

@coderabbitai coderabbitai Bot Jul 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Missing test coverage for negative collectionID rejection.

IngestSqliteWithCollection rejects collectionID < 0 (see ingest.go's validation), but no test in this file exercises that path. Consider adding a case asserting the error for a negative collectionID to cover this critical validation branch.

As per path instructions, tests should "Strive for high test coverage on critical logic paths."

🧰 Tools
🪛 ast-grep (0.44.0)

[warning] 450-450: Detected use of the 'crypto/md5' package (md5.New / md5.Sum). MD5 is a cryptographically broken, collision-prone hash and must not be used for security purposes such as integrity checks, signatures, or password hashing. Use a strong hash from 'crypto/sha256' (sha256.New / sha256.Sum256) or 'crypto/sha512' instead; for passwords use a dedicated KDF such as golang.org/x/crypto/bcrypt or argon2.
Context: md5.Sum
Note: [CWE-327] Use of a Broken or Risky Cryptographic Algorithm.

(weak-hash-md5-go)

🪛 OpenGrep (1.23.0)

[WARNING] 451-451: MD5 and SHA1 are cryptographically broken and should not be used for security purposes. Use SHA-256 or stronger.

(coderabbit.crypto.go-weak-hash)

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/duckdb_parser/ingest_sqlite_test.go` around lines 406 - 523, Add a
focused negative test for IngestSqliteWithCollection that passes a collectionID
below zero and asserts the call returns an error. Reuse the existing setup
helpers in ingest_sqlite_test.go (setupTestParser, createTestSQLite) and target
the IngestSqliteWithCollection method so the validation branch from ingest.go is
covered; keep the assertion specific to the invalid collectionID path and do not
allow ingestion to proceed.

Source: Path instructions

20 changes: 10 additions & 10 deletions pkg/duckdb_parser/templates/ingest_sqlite.go.tmpl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ Key transformations:
ATTACH '{{.FilePath}}' AS src (TYPE sqlite);

INSERT INTO vinfo (
"VM ID", "VM", "Folder ID", "Folder", "Host", "SMBIOS UUID", "VM UUID",
"VM ID", {{if .CollectionID}}"vmmoid", "collection_id", {{end}}"VM", "Folder ID", "Folder", "Host", "SMBIOS UUID", "VM UUID",
"Firmware", "Powerstate", "Connection state", "FT State",
"CPUs", "Memory",
"OS according to the configuration file", "OS according to the VMware Tools",
Expand All @@ -29,7 +29,7 @@ INSERT INTO vinfo (
"migration_excluded", "labels", "guest_apps"
)
SELECT
v.ID,
{{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID), v.ID, {{.CollectionID}},{{else}}v.ID,{{end}}
v.Name,
v.Folder,
v.Folder,
Expand Down Expand Up @@ -70,29 +70,29 @@ WHERE v.IsTemplate = 0;

INSERT INTO vcpu ("VM ID", "Hot Add", "Hot Remove", "Sockets", "Cores p/s")
SELECT
v.ID,
{{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}},
v.CpuHotAddEnabled = 1,
v.CpuHotRemoveEnabled = 1,
v.CpuCount / NULLIF(v.CoresPerSocket, 0),
v.CoresPerSocket
FROM src.VM v
WHERE v.ID IN (SELECT "VM ID" FROM vinfo);
WHERE v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}});

@coderabbitai coderabbitai Bot Jul 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win

Scope related-table filters to the current collection.

Under collection mode these subqueries match any historical vinfo.vmmoid. If vinfo already contains another collection, a current source VM can pass this filter without a matching vinfo row for the current collection_id, producing orphaned hashed rows in vcpu, vmemory, vdisk, or vnetwork.

🔧 Proposed fix 
-WHERE v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}});
+WHERE {{if .CollectionID}}EXISTS (
+    SELECT 1 FROM vinfo
+    WHERE vinfo.vmmoid = v.ID
+      AND vinfo.collection_id = {{.CollectionID}}
+){{else}}v.ID IN (SELECT "VM ID" FROM vinfo){{end}};

Apply the same EXISTS shape to the vmemory, vdisk, and nic_flat filters.

Also applies to: 87-87, 112-112, 125-125

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@pkg/duckdb_parser/templates/ingest_sqlite.go.tmpl` at line 79, The
collection-mode filters in the ingest SQL template are too loose because the
current subqueries on vinfo can match any historical vmmoid and let unrelated
source VMs through. Update the related-table predicates in the template so they
are scoped to the current collection using the same EXISTS pattern, and apply
the change consistently in the vmemory, vdisk, and nic_flat filter blocks as
well as the vcpu case. Use the existing vinfo/vmmoid logic and keep the
collection_id check aligned with the surrounding ingest_sqlite.go.tmpl
conditions.


INSERT INTO vmemory ("VM ID", "Hot Add", "Ballooned")
SELECT
v.ID,
{{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}},
v.MemoryHotAddEnabled = 1,
v.BalloonedMemory
FROM src.VM v
WHERE v.ID IN (SELECT "VM ID" FROM vinfo);
WHERE v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}});

INSERT INTO vdisk (
"VM ID", "Disk Key", "Unit #", "Path", "Disk Path", "Capacity MiB",
"Sharing mode", "Raw", "Shared Bus", "Disk Mode", "Disk UUID",
"Thin", "Controller", "Label", "SCSI Unit #"
)
SELECT
v.ID,
{{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}},
disk->>'key',
disk->>'unitNumber',
disk->>'file',
Expand All @@ -109,11 +109,11 @@ SELECT
disk->>'unitNumber'
FROM src.VM v,
LATERAL unnest(from_json(v.Disks, '[{"key":"INTEGER","unitNumber":"INTEGER","file":"VARCHAR","capacity":"BIGINT","shared":"BOOLEAN","rdm":"BOOLEAN","bus":"VARCHAR","mode":"VARCHAR","serial":"VARCHAR"}]')) AS t(disk)
WHERE v.Disks != '[]' AND v.ID IN (SELECT "VM ID" FROM vinfo);
WHERE v.Disks != '[]' AND v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}});

WITH nic_flat AS (
SELECT
v.ID AS vm_id,
{{if .CollectionID}}md5(CAST({{.CollectionID}} AS VARCHAR) || '_' || v.ID){{else}}v.ID{{end}} AS vm_id,
v.IpAddress,
v.GuestNetworks,
nic.mac AS mac_addr,
Expand All @@ -122,7 +122,7 @@ WITH nic_flat AS (
FROM src.VM v
LEFT JOIN src.Host h ON v.Host = h.ID,
LATERAL unnest(from_json(v.NICs, '[{"network":{"kind":"VARCHAR","id":"VARCHAR"},"mac":"VARCHAR","order":"INTEGER","deviceKey":"INTEGER"}]')) AS t(nic)
WHERE v.NICs != '[]' AND v.ID IN (SELECT "VM ID" FROM vinfo)
WHERE v.NICs != '[]' AND v.ID IN (SELECT {{if .CollectionID}}vmmoid{{else}}"VM ID"{{end}} FROM vinfo{{if .CollectionID}} WHERE vmmoid IS NOT NULL{{end}})
),
gn_flat AS (
SELECT
Expand Down
Loading