Add postgres_synced_tables bundle resource

NEXT_CHANGELOG.md

@@ -13,6 +13,7 @@
 * Make sure warnings asking for approval are understood by agents ([#5239](https://github.com/databricks/cli/pull/5239))
 * Support `replace_existing: true` on `postgres_branches` and `postgres_endpoints` so bundles can manage the implicitly-created production branch and primary read-write endpoint of a Lakebase project.
 * Add `postgres_catalogs` resource to bind a Unity Catalog catalog to a Postgres database on a Lakebase Autoscaling branch ([#5265](https://github.com/databricks/cli/pull/5265)).
+* Add `postgres_synced_tables` resource to sync a Unity Catalog Delta table into a Postgres table on a Lakebase Autoscaling branch ([#5268](https://github.com/databricks/cli/pull/5268)).
 * engine/direct: Changes to state file now persisted to .wal file right away instead of being saved in the end ([#5149](https://github.com/databricks/cli/pull/5149))
 
 ### Dependency updates

acceptance/bundle/invariant/configs/postgres_synced_table.yml.tmpl

@@ -0,0 +1,16 @@
+bundle:
+  name: test-bundle-$UNIQUE_NAME
+
+resources:
+  postgres_synced_tables:
+    foo:
+      synced_table_id: lakebase_$UNIQUE_NAME.public.trips_synced
+      source_table_full_name: main.raw.trips
+      primary_key_columns: [id]
+      scheduling_policy: SNAPSHOT
+      postgres_database: appdb
+      branch: projects/test-pg-project-$UNIQUE_NAME/branches/production
+      create_database_objects_if_missing: true
+      new_pipeline_spec:
+        storage_catalog: main
+        storage_schema: pipelines

acceptance/bundle/invariant/test.toml

@@ -47,6 +47,7 @@ EnvMatrix.INPUT_CONFIG = [
     "postgres_catalog.yml.tmpl",
     "postgres_endpoint.yml.tmpl",
     "postgres_project.yml.tmpl",
+    "postgres_synced_table.yml.tmpl",
     "registered_model.yml.tmpl",
     "schema.yml.tmpl",
     "schema_grant_ref.yml.tmpl",
@@ -69,6 +70,7 @@ no_postgres_project_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_proj
 no_postgres_branch_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_branch.yml.tmpl"]
 no_postgres_endpoint_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_endpoint.yml.tmpl"]
 no_postgres_catalog_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_catalog.yml.tmpl"]
+no_postgres_synced_table_on_cloud = ["CONFIG_Cloud=true", "INPUT_CONFIG=postgres_synced_table.yml.tmpl"]
 
 # External locations require actual storage credentials with cloud IAM setup
 # which are environment-specific, so we only test locally with the mock server

acceptance/bundle/refschema/out.fields.txt

@@ -2850,6 +2850,49 @@ resources.postgres_projects.*.permissions[*].group_name	string	ALL
 resources.postgres_projects.*.permissions[*].level	iam.PermissionLevel	ALL
 resources.postgres_projects.*.permissions[*].service_principal_name	string	ALL
 resources.postgres_projects.*.permissions[*].user_name	string	ALL
+resources.postgres_synced_tables.*.branch	string	ALL
+resources.postgres_synced_tables.*.create_database_objects_if_missing	bool	ALL
+resources.postgres_synced_tables.*.create_time	*time.Time	REMOTE
+resources.postgres_synced_tables.*.existing_pipeline_id	string	ALL
+resources.postgres_synced_tables.*.id	string	INPUT
+resources.postgres_synced_tables.*.lifecycle	resources.Lifecycle	INPUT
+resources.postgres_synced_tables.*.lifecycle.prevent_destroy	bool	INPUT
+resources.postgres_synced_tables.*.modified_status	string	INPUT
+resources.postgres_synced_tables.*.name	string	REMOTE
+resources.postgres_synced_tables.*.new_pipeline_spec	*postgres.NewPipelineSpec	ALL
+resources.postgres_synced_tables.*.new_pipeline_spec.budget_policy_id	string	ALL
+resources.postgres_synced_tables.*.new_pipeline_spec.storage_catalog	string	ALL
+resources.postgres_synced_tables.*.new_pipeline_spec.storage_schema	string	ALL
+resources.postgres_synced_tables.*.postgres_database	string	ALL
+resources.postgres_synced_tables.*.primary_key_columns	[]string	ALL
+resources.postgres_synced_tables.*.primary_key_columns[*]	string	ALL
+resources.postgres_synced_tables.*.scheduling_policy	postgres.SyncedTableSyncedTableSpecSyncedTableSchedulingPolicy	ALL
+resources.postgres_synced_tables.*.source_table_full_name	string	ALL
+resources.postgres_synced_tables.*.status	*postgres.SyncedTableSyncedTableStatus	REMOTE
+resources.postgres_synced_tables.*.status.detailed_state	postgres.SyncedTableState	REMOTE
+resources.postgres_synced_tables.*.status.last_processed_commit_version	int64	REMOTE
+resources.postgres_synced_tables.*.status.last_sync	*postgres.SyncedTablePosition	REMOTE
+resources.postgres_synced_tables.*.status.last_sync.delta_table_sync_info	*postgres.DeltaTableSyncInfo	REMOTE
+resources.postgres_synced_tables.*.status.last_sync.delta_table_sync_info.delta_commit_time	*time.Time	REMOTE
+resources.postgres_synced_tables.*.status.last_sync.delta_table_sync_info.delta_commit_version	int64	REMOTE
+resources.postgres_synced_tables.*.status.last_sync.sync_end_time	*time.Time	REMOTE
+resources.postgres_synced_tables.*.status.last_sync.sync_start_time	*time.Time	REMOTE
+resources.postgres_synced_tables.*.status.last_sync_time	*time.Time	REMOTE
+resources.postgres_synced_tables.*.status.message	string	REMOTE
+resources.postgres_synced_tables.*.status.ongoing_sync_progress	*postgres.SyncedTablePipelineProgress	REMOTE
+resources.postgres_synced_tables.*.status.ongoing_sync_progress.estimated_completion_time_seconds	float64	REMOTE
+resources.postgres_synced_tables.*.status.ongoing_sync_progress.latest_version_currently_processing	int64	REMOTE
+resources.postgres_synced_tables.*.status.ongoing_sync_progress.sync_progress_completion	float64	REMOTE
+resources.postgres_synced_tables.*.status.ongoing_sync_progress.synced_row_count	int64	REMOTE
+resources.postgres_synced_tables.*.status.ongoing_sync_progress.total_row_count	int64	REMOTE
+resources.postgres_synced_tables.*.status.pipeline_id	string	REMOTE
+resources.postgres_synced_tables.*.status.project	string	REMOTE
+resources.postgres_synced_tables.*.status.provisioning_phase	postgres.ProvisioningPhase	REMOTE
+resources.postgres_synced_tables.*.status.unity_catalog_provisioning_state	postgres.ProvisioningInfoState	REMOTE
+resources.postgres_synced_tables.*.synced_table_id	string	ALL
+resources.postgres_synced_tables.*.timeseries_key	string	ALL
+resources.postgres_synced_tables.*.uid	string	REMOTE
+resources.postgres_synced_tables.*.url	string	INPUT
 resources.quality_monitors.*.assets_dir	string	ALL
 resources.quality_monitors.*.baseline_table_name	string	ALL
 resources.quality_monitors.*.custom_metrics	[]catalog.MonitorMetric	ALL

acceptance/bundle/resources/postgres_synced_tables/basic/databricks.yml.tmpl

@@ -0,0 +1,38 @@
+bundle:
+  name: deploy-postgres-synced-table-$UNIQUE_NAME
+
+sync:
+  paths: []
+
+resources:
+  schemas:
+    pipeline_storage:
+      name: pipeline_storage_$UNIQUE_NAME
+      catalog_name: main
+      comment: "Pipeline storage for the synced-table test"
+
+  postgres_projects:
+    my_project:
+      project_id: test-pg-proj-$UNIQUE_NAME
+      display_name: "Test Project for Synced Table"
+      pg_version: 17
+
+  postgres_catalogs:
+    my_catalog:
+      catalog_id: lakebase_test_$UNIQUE_NAME
+      branch: ${resources.postgres_projects.my_project.id}/branches/production
+      postgres_database: appdb
+      create_database_if_missing: true
+
+  postgres_synced_tables:
+    my_table:
+      synced_table_id: ${resources.postgres_catalogs.my_catalog.catalog_id}.public.trips_synced
+      source_table_full_name: main.source_$UNIQUE_NAME.trips_source
+      primary_key_columns: ["tpep_pickup_datetime"]
+      scheduling_policy: SNAPSHOT
+      postgres_database: appdb
+      branch: ${resources.postgres_projects.my_project.id}/branches/production
+      create_database_objects_if_missing: true
+      new_pipeline_spec:
+        storage_catalog: ${resources.schemas.pipeline_storage.catalog_name}
+        storage_schema: ${resources.schemas.pipeline_storage.name}

acceptance/bundle/resources/postgres_synced_tables/basic/out.requests.json

@@ -0,0 +1,54 @@
+{
+  "method": "POST",
+  "path": "/api/2.0/postgres/projects",
+  "q": {
+    "project_id": "test-pg-proj-[UNIQUE_NAME]"
+  },
+  "body": {
+    "spec": {
+      "display_name": "Test Project for Synced Table",
+      "pg_version": 17
+    }
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/postgres/catalogs",
+  "q": {
+    "catalog_id": "lakebase_test_[UNIQUE_NAME]"
+  },
+  "body": {
+    "spec": {
+      "branch": "projects/test-pg-proj-[UNIQUE_NAME]/branches/production",
+      "create_database_if_missing": true,
+      "postgres_database": "appdb"
+    }
+  }
+}
+{
+  "method": "POST",
+  "path": "/api/2.0/postgres/synced_tables",
+  "q": {
+    "synced_table_id": "lakebase_test_[UNIQUE_NAME].public.trips_synced"
+  },
+  "body": {
+    "spec": {
+      "branch": "projects/test-pg-proj-[UNIQUE_NAME]/branches/production",
+      "create_database_objects_if_missing": true,
+      "new_pipeline_spec": {
+        "storage_catalog": "main",
+        "storage_schema": "pipeline_storage_[UNIQUE_NAME]"
+      },
+      "postgres_database": "appdb",
+      "primary_key_columns": [
+        "tpep_pickup_datetime"
+      ],
+      "scheduling_policy": "SNAPSHOT",
+      "source_table_full_name": "main.source_[UNIQUE_NAME].trips_source"
+    }
+  }
+}
+{
+  "method": "GET",
+  "path": "/api/2.0/postgres/synced_tables/lakebase_test_[UNIQUE_NAME].public.trips_synced"
+}

acceptance/bundle/resources/postgres_synced_tables/basic/output.txt

@@ -0,0 +1,95 @@
+Creating temporary source table: main.source_[UNIQUE_NAME].trips_source
+{
+  "full_name": "main.source_[UNIQUE_NAME]"
+}
+
+>>> [CLI] bundle validate
+Name: deploy-postgres-synced-table-[UNIQUE_NAME]
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/deploy-postgres-synced-table-[UNIQUE_NAME]/default
+
+Validation OK!
+
+>>> [CLI] bundle summary
+Name: deploy-postgres-synced-table-[UNIQUE_NAME]
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/deploy-postgres-synced-table-[UNIQUE_NAME]/default
+Resources:
+  Postgres catalogs:
+    my_catalog:
+      Name: lakebase_test_[UNIQUE_NAME]
+      URL:  [DATABRICKS_URL]/explore/data/lakebase_test_[UNIQUE_NAME]
+  Postgres projects:
+    my_project:
+      Name: Test Project for Synced Table
+      URL:  (not deployed)
+  Postgres synced tables:
+    my_table:
+      Name: ${resources.postgres_catalogs.my_catalog.catalog_id}.public.trips_synced
+      URL:  [DATABRICKS_URL]/explore/data/$%7Bresources/postgres_catalogs/my_catalog.catalog_id%7D.public.trips_synced
+  Schemas:
+    pipeline_storage:
+      Name: pipeline_storage_[UNIQUE_NAME]
+      URL:  (not deployed)
+
+>>> [CLI] bundle deploy
+Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/deploy-postgres-synced-table-[UNIQUE_NAME]/default/files...
+Deploying resources...
+Updating deployment state...
+Deployment complete!
+
+>>> [CLI] postgres get-synced-table synced_tables/lakebase_test_[UNIQUE_NAME].public.trips_synced
+{
+  "name": "synced_tables/lakebase_test_[UNIQUE_NAME].public.trips_synced",
+  "unity_catalog_provisioning_state": "ACTIVE"
+}
+
+>>> [CLI] bundle summary
+Name: deploy-postgres-synced-table-[UNIQUE_NAME]
+Target: default
+Workspace:
+  User: [USERNAME]
+  Path: /Workspace/Users/[USERNAME]/.bundle/deploy-postgres-synced-table-[UNIQUE_NAME]/default
+Resources:
+  Postgres catalogs:
+    my_catalog:
+      Name: lakebase_test_[UNIQUE_NAME]
+      URL:  [DATABRICKS_URL]/explore/data/lakebase_test_[UNIQUE_NAME]
+  Postgres projects:
+    my_project:
+      Name: Test Project for Synced Table
+      URL:  (not deployed)
+  Postgres synced tables:
+    my_table:
+      Name: lakebase_test_[UNIQUE_NAME].public.trips_synced
+      URL:  [DATABRICKS_URL]/explore/data/lakebase_test_[UNIQUE_NAME]/public/trips_synced
+  Schemas:
+    pipeline_storage:
+      Name: pipeline_storage_[UNIQUE_NAME]
+      URL:  [DATABRICKS_URL]/explore/data/main/pipeline_storage_[UNIQUE_NAME]
+
+>>> print_requests.py --keep --get //postgres ^//workspace-files/ ^//workspace/ ^//telemetry-ext ^//operations/
+
+>>> [CLI] bundle destroy --auto-approve
+The following resources will be deleted:
+  delete resources.postgres_catalogs.my_catalog
+  delete resources.postgres_projects.my_project
+  delete resources.postgres_synced_tables.my_table
+  delete resources.schemas.pipeline_storage
+
+This action will result in the deletion of the following UC schemas. Any underlying data may be lost:
+  delete resources.schemas.pipeline_storage
+
+This action will result in the deletion of the following Lakebase projects along with
+all their branches, databases, and endpoints. All data stored in them will be permanently lost:
+  delete resources.postgres_projects.my_project
+
+All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/deploy-postgres-synced-table-[UNIQUE_NAME]/default
+
+Deleting files...
+Destroy complete!
+Cleaning up temporary source table

acceptance/bundle/resources/postgres_synced_tables/basic/script

@@ -0,0 +1,39 @@
+envsubst < databricks.yml.tmpl > databricks.yml
+
+# Create a per-test source schema + table. We can't read samples.nyctaxi.trips
+# directly because shared workspaces hit the "20 synced tables per source table"
+# limit (same reason synced_database_tables takes this approach).
+echo "Creating temporary source table: main.source_$UNIQUE_NAME.trips_source"
+$CLI schemas create source_$UNIQUE_NAME main -o json | jq '{full_name}'
+MSYS_NO_PATHCONV=1 $CLI api post "/api/2.0/sql/statements/" --json "{
+    \"warehouse_id\": \"$TEST_DEFAULT_WAREHOUSE_ID\",
+    \"statement\": \"CREATE TABLE main.source_$UNIQUE_NAME.trips_source AS SELECT * FROM samples.nyctaxi.trips LIMIT 10\",
+    \"wait_timeout\": \"45s\"
+  }" > /dev/null
+
+cleanup() {
+    trace $CLI bundle destroy --auto-approve
+    echo "Cleaning up temporary source table"
+    $CLI tables delete main.source_$UNIQUE_NAME.trips_source || true
+    $CLI schemas delete main.source_$UNIQUE_NAME || true
+    rm -f out.requests.txt
+}
+trap cleanup EXIT
+
+trace $CLI bundle validate
+
+trace $CLI bundle summary
+
+rm -f out.requests.txt
+trace $CLI bundle deploy
+
+# Keep only the deterministic identity + provisioning-state. detailed_state
+# varies with pipeline-provisioning timing on cloud, ongoing_sync_progress and
+# project only appear there, and create_time/update_time/uid/pipeline_id are
+# random per run.
+trace $CLI postgres get-synced-table "synced_tables/lakebase_test_${UNIQUE_NAME}.public.trips_synced" | jq '{name, unity_catalog_provisioning_state: .status.unity_catalog_provisioning_state}'
+
+trace $CLI bundle summary
+
+# Filter requests to only show postgres operations (exclude workspace, telemetry, and operation polling).
+trace print_requests.py --keep --get '//postgres' '^//workspace-files/' '^//workspace/' '^//telemetry-ext' '^//operations/' > out.requests.json

acceptance/bundle/resources/postgres_synced_tables/basic/test.toml

@@ -0,0 +1 @@
+# All configuration inherited from parent test.toml

acceptance/bundle/resources/postgres_synced_tables/recreate/databricks.yml.tmpl

@@ -0,0 +1,39 @@
+bundle:
+  name: recreate-postgres-synced-table-$UNIQUE_NAME
+
+sync:
+  paths: []
+
+resources:
+  schemas:
+    pipeline_storage:
+      name: pipeline_storage_$UNIQUE_NAME
+      catalog_name: main
+      comment: "Pipeline storage for the synced-table recreate test"
+
+  postgres_projects:
+    my_project:
+      project_id: test-pg-proj-$UNIQUE_NAME
+      display_name: "Test Project for Synced Table Recreate"
+      pg_version: 17
+
+  postgres_catalogs:
+    my_catalog:
+      catalog_id: lakebase_test_$UNIQUE_NAME
+      branch: ${resources.postgres_projects.my_project.id}/branches/production
+      postgres_database: appdb
+      create_database_if_missing: true
+
+  postgres_synced_tables:
+    my_table:
+      synced_table_id: ${resources.postgres_catalogs.my_catalog.catalog_id}.public.trips_synced
+      source_table_full_name: main.source_$UNIQUE_NAME.trips_source
+      primary_key_columns: ["tpep_pickup_datetime"]
+      scheduling_policy: SNAPSHOT
+      postgres_database: appdb
+      branch: ${resources.postgres_projects.my_project.id}/branches/production
+      create_database_objects_if_missing: true
+      timeseries_key: $TIMESERIES_KEY
+      new_pipeline_spec:
+        storage_catalog: ${resources.schemas.pipeline_storage.catalog_name}
+        storage_schema: ${resources.schemas.pipeline_storage.name}