feat(control-plane): version data import table naming

README.md

@@ -952,6 +952,7 @@ The shared K8s pool spawns workers on-demand, reserves them per org, activates t
 Managed-warehouse contract notes:
 
 - At most one managed-warehouse row exists per team. The row may be absent before first provisioning or after cleanup, but there is never more than one active warehouse contract for a team.
+- Each org has a `data_imports_table_naming_version`. Migration `000034` assigns `legacy_batch_v1` to orgs that already exist and changes the database default to `copy_v1` for orgs created afterward. `GET /api/v1/orgs/:id/teams` returns the org-level value alongside the team rows so data-import writers can pin one physical table name consistently.
 - The admin API exposes that contract at `GET /api/v1/teams/:name/warehouse` and `PUT /api/v1/teams/:name/warehouse`. Team list/get responses also include a nested `warehouse` object when present.
 - Org rows support optional `max_vcpus` on `POST /api/v1/orgs` and `PUT /api/v1/orgs/:id`. In K8s multi-tenant mode, this caps the org's active admitted worker pod vCPUs; `0` means unlimited.
 - User rows support an optional `max_vcpus` field on `POST /api/v1/users` and `PUT /api/v1/orgs/:id/users/:username`. `max_vcpus` limits the user's active admitted worker pod vCPUs in K8s multi-tenant mode; `0` means unlimited.

controlplane/configstore/data_imports_table_naming.go

@@ -0,0 +1,6 @@
+package configstore
+
+const (
+	DataImportsTableNamingVersionLegacyBatchV1 = "legacy_batch_v1"
+	DataImportsTableNamingVersionCopyV1        = "copy_v1"
+)

controlplane/configstore/migrations/000034_org_data_imports_table_naming.sql

@@ -0,0 +1,17 @@
+-- +goose Up
+ALTER TABLE duckgres_orgs
+    ADD COLUMN IF NOT EXISTS data_imports_table_naming_version VARCHAR(32) NOT NULL DEFAULT 'legacy_batch_v1';
+
+ALTER TABLE duckgres_orgs
+    ALTER COLUMN data_imports_table_naming_version SET DEFAULT 'copy_v1';
+
+ALTER TABLE duckgres_orgs
+    ADD CONSTRAINT duckgres_orgs_data_imports_table_naming_version_check
+    CHECK (data_imports_table_naming_version IN ('legacy_batch_v1', 'copy_v1'));
+
+-- +goose Down
+ALTER TABLE duckgres_orgs
+    DROP CONSTRAINT IF EXISTS duckgres_orgs_data_imports_table_naming_version_check;
+
+ALTER TABLE duckgres_orgs
+    DROP COLUMN IF EXISTS data_imports_table_naming_version;

controlplane/configstore/models.go

@@ -21,14 +21,15 @@ type Org struct {
 	// human editability) applied to connections that don't size themselves via
 	// the duckgres.worker_* startup options. Empty = unset. Versioned SQL
 	// migrations add these columns.
-	DefaultWorkerCPU        string `gorm:"size:32" json:"default_worker_cpu"`
-	DefaultWorkerMemory     string `gorm:"size:32" json:"default_worker_memory"`
-	DefaultWorkerTTL        string `gorm:"size:32" json:"default_worker_ttl"`
-	DefaultWorkerMinHotIdle int    `gorm:"default:0" json:"default_worker_min_hot_idle"`
-	Teams         []OrgTeam         `gorm:"foreignKey:OrgID;references:Name;constraint:OnDelete:CASCADE" json:"teams,omitempty"`
-	Users         []OrgUser         `gorm:"foreignKey:OrgID;references:Name" json:"users,omitempty"`
-	Warehouse     *ManagedWarehouse `gorm:"foreignKey:OrgID;references:Name;constraint:OnDelete:CASCADE" json:"warehouse,omitempty"`
-	CreatedAt     time.Time         `json:"created_at"`
+	DefaultWorkerCPU              string            `gorm:"size:32" json:"default_worker_cpu"`
+	DefaultWorkerMemory           string            `gorm:"size:32" json:"default_worker_memory"`
+	DefaultWorkerTTL              string            `gorm:"size:32" json:"default_worker_ttl"`
+	DefaultWorkerMinHotIdle       int               `gorm:"default:0" json:"default_worker_min_hot_idle"`
+	DataImportsTableNamingVersion string            `gorm:"size:32;not null;default:copy_v1" json:"data_imports_table_naming_version"`
+	Teams                         []OrgTeam         `gorm:"foreignKey:OrgID;references:Name;constraint:OnDelete:CASCADE" json:"teams,omitempty"`
+	Users                         []OrgUser         `gorm:"foreignKey:OrgID;references:Name" json:"users,omitempty"`
+	Warehouse                     *ManagedWarehouse `gorm:"foreignKey:OrgID;references:Name;constraint:OnDelete:CASCADE" json:"warehouse,omitempty"`
+	CreatedAt                     time.Time         `json:"created_at"`
 	// UpdatedAt doubles as an input to the discovery change marker
 	// (ConfigStore.LatestConfigChange): DeleteOrgTeamTx touches it so a
 	// team-row DELETE — which leaves no updated_at of its own behind —

controlplane/provisioning/api.go

@@ -602,6 +602,15 @@ type orgTeamUpsertRequest struct {
 
 func (h *handler) listOrgTeams(c *gin.Context) {
 	orgID := c.Param("id")
+	org, err := h.store.GetOrg(orgID)
+	if err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			c.JSON(http.StatusNotFound, gin.H{"error": "org not found"})
+			return
+		}
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
 	teams, err := h.store.ListOrgTeams(orgID)
 	if err != nil {
 		if errors.Is(err, gorm.ErrRecordNotFound) {
@@ -611,7 +620,10 @@ func (h *handler) listOrgTeams(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
 	}
-	c.JSON(http.StatusOK, gin.H{"teams": teams})
+	c.JSON(http.StatusOK, gin.H{
+		"teams":                             teams,
+		"data_imports_table_naming_version": org.DataImportsTableNamingVersion,
+	})
 }
 
 // upsertOrgTeam creates or overwrites one (org, team) row. This endpoint IS

controlplane/provisioning/api_test.go

@@ -1174,7 +1174,10 @@ func TestOrgTeamUpsertValidation(t *testing.T) {
 
 func TestOrgTeamUpsertCreatesAndLists(t *testing.T) {
 	store := newFakeStore()
-	store.orgs["acme"] = &configstore.Org{Name: "acme"}
+	store.orgs["acme"] = &configstore.Org{
+		Name:                          "acme",
+		DataImportsTableNamingVersion: configstore.DataImportsTableNamingVersionCopyV1,
+	}
 	router := newTestRouter(store)
 
 	rec := doJSON(t, router, http.MethodPost, "/api/v1/orgs/acme/teams",
@@ -1198,14 +1201,18 @@ func TestOrgTeamUpsertCreatesAndLists(t *testing.T) {
 		t.Fatalf("list status = %d, want 200: %s", rec.Code, rec.Body.String())
 	}
 	var listing struct {
-		Teams []configstore.OrgTeam `json:"teams"`
+		Teams                         []configstore.OrgTeam `json:"teams"`
+		DataImportsTableNamingVersion string                `json:"data_imports_table_naming_version"`
 	}
 	if err := json.Unmarshal(rec.Body.Bytes(), &listing); err != nil {
 		t.Fatalf("decode listing: %v", err)
 	}
 	if len(listing.Teams) != 1 || listing.Teams[0].TeamID != 7 {
 		t.Fatalf("listing = %+v, want the created team", listing.Teams)
 	}
+	if listing.DataImportsTableNamingVersion != configstore.DataImportsTableNamingVersionCopyV1 {
+		t.Fatalf("data imports naming version = %q, want copy_v1", listing.DataImportsTableNamingVersion)
+	}
 }
 
 func TestOrgTeamListUnknownOrg404(t *testing.T) {

tests/configstore/migrations_postgres_test.go

@@ -52,7 +52,8 @@ func TestConfigStoreRunsVersionedSQLMigrations(t *testing.T) {
 	requireGooseMigrationRecorded(t, db, 31)
 	requireGooseMigrationRecorded(t, db, 32)
 	requireGooseMigrationRecorded(t, db, 33)
-	requireGooseLatestVersion(t, db, 33)
+	requireGooseMigrationRecorded(t, db, 34)
+	requireGooseLatestVersion(t, db, 34)
 	requireTableAbsent(t, db, "duckgres_schema_migrations")
 
 	// Migration 000018 added the reshard operation + verbose log tables.
@@ -120,6 +121,11 @@ func TestConfigStoreRunsVersionedSQLMigrations(t *testing.T) {
 	requireColumnNullable(t, db, "duckgres_org_teams", "schema_data_imports_name")
 	requireUniqueIndex(t, db, "duckgres_org_teams", "org_id,schema_name")
 
+	// Migration 000034 preserves the table naming used by existing orgs while
+	// selecting the copy workflow naming for orgs created after deployment.
+	requireColumnNotNull(t, db, "duckgres_orgs", "data_imports_table_naming_version")
+	requireColumnDefault(t, db, "duckgres_orgs", "data_imports_table_naming_version", "'copy_v1'::character varying")
+
 	// Migration 000026 added PostHog's cached earliest-event date (nullable
 	// DATE — NULL until the PostHog sensor resolves it).
 	requireColumnNullable(t, db, "duckgres_org_teams", "earliest_event_date")
@@ -207,8 +213,8 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) {
 	t.Cleanup(func() {
 		_ = baselineDB.Close()
 	})
-
 	if err := store.DB().Exec(`
+			ALTER TABLE duckgres_orgs DROP COLUMN data_imports_table_naming_version;
 			ALTER TABLE duckgres_orgs DROP COLUMN max_vcpus;
 			ALTER TABLE duckgres_org_users DROP COLUMN max_vcpus;
 			ALTER TABLE duckgres_org_users DROP COLUMN disabled;
@@ -237,7 +243,7 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) {
 			);
 			DROP TABLE IF EXISTS duckgres_reshard_operation_log;
 			DROP TABLE IF EXISTS duckgres_reshard_operations;
-			DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33);
+			DELETE FROM goose_db_version WHERE version_id IN (9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34);
 		`).Error; err != nil {
 		t.Fatalf("downgrade baseline schema to pre-v9 shape: %v", err)
 	}
@@ -284,7 +290,8 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) {
 	requireGooseMigrationRecorded(t, upgradedDB, 31)
 	requireGooseMigrationRecorded(t, upgradedDB, 32)
 	requireGooseMigrationRecorded(t, upgradedDB, 33)
-	requireGooseLatestVersion(t, upgradedDB, 33)
+	requireGooseMigrationRecorded(t, upgradedDB, 34)
+	requireGooseLatestVersion(t, upgradedDB, 34)
 	requireColumnPresent(t, upgradedDB, "duckgres_reshard_operations", "password_url")
 	requireTablePresent(t, upgradedDB, "duckgres_worker_spawn_log")
 	requireColumnDefault(t, upgradedDB, "duckgres_orgs", "max_vcpus", "0")
@@ -300,6 +307,68 @@ func TestConfigStoreSQLMigrationsUpgradeVersion8Schema(t *testing.T) {
 	requireColumnAbsent(t, upgradedDB, "duckgres_managed_warehouses", "iceberg_enabled")
 	requireColumnDefault(t, upgradedDB, "duckgres_managed_warehouses", "metadata_proxy_enabled", "false")
 	requireColumnAbsent(t, upgradedDB, "duckgres_org_users", "default_catalog")
+
+}
+
+func TestConfigStoreSQLMigration34VersionsExistingAndNewOrgs(t *testing.T) {
+	_, connStr := newIsolatedConfigStoreSchema(t)
+	store, err := cpconfigStoreNew(connStr)
+	if err != nil {
+		t.Fatalf("create baseline config store: %v", err)
+	}
+	baselineDB := storeDB(t, store)
+	t.Cleanup(func() {
+		_ = baselineDB.Close()
+	})
+
+	if err := store.DB().Exec(`
+		INSERT INTO duckgres_orgs (name, database_name, created_at, updated_at)
+		VALUES ('existing-naming-policy', 'existing-naming-policy', now(), now());
+		ALTER TABLE duckgres_orgs DROP COLUMN data_imports_table_naming_version;
+		DELETE FROM goose_db_version WHERE version_id = 34;
+	`).Error; err != nil {
+		t.Fatalf("restore pre-migration-34 schema: %v", err)
+	}
+	requireGooseLatestVersion(t, baselineDB, 33)
+
+	upgradedStore, err := cpconfigStoreNew(connStr)
+	if err != nil {
+		t.Fatalf("apply migration 34: %v", err)
+	}
+	upgradedDB := storeDB(t, upgradedStore)
+	t.Cleanup(func() {
+		_ = upgradedDB.Close()
+	})
+
+	var existingNamingVersion string
+	if err := upgradedStore.DB().Raw(`
+		SELECT data_imports_table_naming_version
+		FROM duckgres_orgs
+		WHERE name = 'existing-naming-policy'
+	`).Scan(&existingNamingVersion).Error; err != nil {
+		t.Fatalf("read existing org naming version: %v", err)
+	}
+	if existingNamingVersion != cpconfigstore.DataImportsTableNamingVersionLegacyBatchV1 {
+		t.Fatalf("existing org naming version = %q, want legacy_batch_v1", existingNamingVersion)
+	}
+
+	if err := upgradedStore.DB().Exec(`
+		INSERT INTO duckgres_orgs (name, database_name, created_at, updated_at)
+		VALUES ('new-naming-policy', 'new-naming-policy', now(), now())
+	`).Error; err != nil {
+		t.Fatalf("create org after naming migration: %v", err)
+	}
+	var newNamingVersion string
+	if err := upgradedStore.DB().Raw(`
+		SELECT data_imports_table_naming_version
+		FROM duckgres_orgs
+		WHERE name = 'new-naming-policy'
+	`).Scan(&newNamingVersion).Error; err != nil {
+		t.Fatalf("read new org naming version: %v", err)
+	}
+	if newNamingVersion != cpconfigstore.DataImportsTableNamingVersionCopyV1 {
+		t.Fatalf("new org naming version = %q, want copy_v1", newNamingVersion)
+	}
 }
 
 func TestConfigStoreSQLMigrationsUpgradeOldOrgSchema(t *testing.T) {
@@ -368,6 +437,17 @@ func TestConfigStoreSQLMigrationsUpgradeOldOrgSchema(t *testing.T) {
 	}
 	requireColumnAbsent(t, sqlDB, "duckgres_orgs", "max_connections")
 	requireGooseMigrationRecorded(t, sqlDB, 3)
+	var namingVersion string
+	if err := store.DB().Raw(`
+		SELECT data_imports_table_naming_version
+		FROM duckgres_orgs
+		WHERE name = 'old-org'
+	`).Scan(&namingVersion).Error; err != nil {
+		t.Fatalf("read migrated data imports naming version: %v", err)
+	}
+	if namingVersion != cpconfigstore.DataImportsTableNamingVersionLegacyBatchV1 {
+		t.Fatalf("migrated data imports naming version = %q, want legacy_batch_v1", namingVersion)
+	}
 
 	// Migration 000024 backfilled the legacy default_team_id value into the
 	// org's team row and dropped the column.