You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2026/09/06 12:00:20 UTC
(terraform-provider-paimon) branch main updated: Align Paimon API contracts and harden provider recovery (#17)
This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/terraform-provider-paimon.git
The following commit(s) were added to refs/heads/main by this push:
new 0cd3f38 Align Paimon API contracts and harden provider recovery (#17)
0cd3f38 is described below
commit 0cd3f38600faad17f6f05c24deedd17583879834
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Sun Sep 6 20:00:16 2026 +0800
Align Paimon API contracts and harden provider recovery (#17)
---
.github/workflows/catalog-integration.yml | 68 +++
.github/workflows/go-ci.yml | 29 ++
Makefile | 14 +-
README.md | 37 +-
dev/release/verify_registry.sh | 72 ++++
docs/api-contract.md | 84 ++++
docs/data-sources/database.md | 41 +-
docs/data-sources/table.md | 62 ++-
docs/index.md | 117 +++--
docs/production-readiness.md | 112 +++++
docs/releasing.md | 21 +
docs/resources/column_mask.md | 56 ++-
docs/resources/database.md | 37 +-
docs/resources/permission.md | 36 ++
docs/resources/row_filter.md | 55 ++-
docs/resources/table.md | 174 +++++++-
examples/main.tf | 4 +-
examples/{ => production}/main.tf | 54 ++-
internal/client/client.go | 108 +++--
internal/client/client_test.go | 100 ++++-
internal/client/dlf_auth.go | 77 +++-
internal/client/dlf_auth_test.go | 68 +++
internal/provider/acceptance_test.go | 34 +-
internal/provider/api_contract_test.go | 504 ++++++++++++++++++++++
internal/provider/data_source_database.go | 10 +-
internal/provider/data_source_table.go | 4 +-
internal/provider/documentation_test.go | 151 +++++++
internal/provider/helpers.go | 24 +-
internal/provider/import_acceptance_test.go | 66 +++
internal/provider/management_acceptance_test.go | 200 +++++++++
internal/provider/policy_semantics.go | 198 +++++++++
internal/provider/production_readiness_test.go | 440 +++++++++++++++++++
internal/provider/provider.go | 87 +++-
internal/provider/provider_management_test.go | 45 +-
internal/provider/provider_test.go | 6 +-
internal/provider/real_catalog_acceptance_test.go | 95 ++++
internal/provider/resource_database.go | 39 +-
internal/provider/resource_permission.go | 4 +-
internal/provider/resource_policy.go | 222 +++++++---
internal/provider/resource_table.go | 195 ++++++---
internal/provider/table_options.go | 72 +++-
internal/provider/table_plan.go | 206 +++++++++
internal/provider/table_schema.go | 78 ++--
43 files changed, 3744 insertions(+), 362 deletions(-)
diff --git a/.github/workflows/catalog-integration.yml b/.github/workflows/catalog-integration.yml
new file mode 100644
index 0000000..3f8ffed
--- /dev/null
+++ b/.github/workflows/catalog-integration.yml
@@ -0,0 +1,68 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+name: Live catalog acceptance
+
+on:
+ workflow_dispatch:
+ inputs:
+ server_version:
+ description: 'Exact tested Paimon REST or DLF server version'
+ required: true
+ type: string
+
+permissions:
+ contents: read
+
+jobs:
+ integration:
+ runs-on: ubuntu-latest
+ environment: catalog-integration
+ timeout-minutes: 30
+ steps:
+ - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
+ with:
+ persist-credentials: false
+ - uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e
+ with:
+ go-version-file: go.mod
+ cache: false
+ - uses: hashicorp/setup-terraform@dfe3c3f87815947d99a8997f908cb6525fc44e9e
+ with:
+ terraform_version: '1.13.5'
+ terraform_wrapper: false
+ - name: Exercise isolated catalog
+ env:
+ PAIMON_ACC_URI: ${{ secrets.PAIMON_ACC_URI }}
+ PAIMON_ACC_ALLOW_MUTATIONS: '1'
+ PAIMON_ACC_PRINCIPAL: ${{ secrets.PAIMON_ACC_PRINCIPAL }}
+ TF_VAR_paimon_warehouse: ${{ secrets.PAIMON_WAREHOUSE }}
+ TF_VAR_paimon_token_provider: ${{ vars.PAIMON_TOKEN_PROVIDER }}
+ TF_VAR_paimon_token: ${{ secrets.PAIMON_TOKEN }}
+ TF_VAR_paimon_dlf_region: ${{ vars.PAIMON_DLF_REGION }}
+ TF_VAR_paimon_dlf_signing_algorithm: ${{ vars.PAIMON_DLF_SIGNING_ALGORITHM }}
+ TF_VAR_paimon_dlf_access_key_id: ${{ secrets.PAIMON_DLF_ACCESS_KEY_ID }}
+ TF_VAR_paimon_dlf_access_key_secret: ${{ secrets.PAIMON_DLF_ACCESS_KEY_SECRET }}
+ TF_VAR_paimon_dlf_security_token: ${{ secrets.PAIMON_DLF_SECURITY_TOKEN }}
+ TESTED_SERVER_VERSION: ${{ inputs.server_version }}
+ run: |
+ test -n "$PAIMON_ACC_PRINCIPAL" || { echo 'Configure an isolated PAIMON_ACC_PRINCIPAL to exercise all management resources' >&2; exit 1; }
+ printf 'Catalog version under test: %s\n' "$TESTED_SERVER_VERSION"
+ for name in ${!TF_VAR_@}; do
+ if [[ -z "${!name}" ]]; then unset "$name"; fi
+ done
+ make test-integration
diff --git a/.github/workflows/go-ci.yml b/.github/workflows/go-ci.yml
index fcaccfb..132dafd 100644
--- a/.github/workflows/go-ci.yml
+++ b/.github/workflows/go-ci.yml
@@ -111,5 +111,34 @@ jobs:
cache-dependency-path: go.sum
- name: Install Terraform
uses: hashicorp/setup-terraform@dfe3c3f87815947d99a8997f908cb6525fc44e9e # v4.0.1
+ with:
+ terraform_version: '1.13.5'
+ terraform_wrapper: false
- name: Run acceptance tests
run: make test-acceptance
+
+ tofu-acceptance:
+ name: OpenTofu protocol acceptance
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+ with:
+ persist-credentials: false
+ - name: Install Go
+ uses: actions/setup-go@b7ad1dad31e06c5925ef5d2fc7ad053ef454303e # v7.0.0
+ with:
+ go-version-file: go.mod
+ cache: false
+ - name: Install OpenTofu
+ env:
+ TOFU_VERSION: '1.12.6'
+ TOFU_SHA256: '5dc43da4f750f33873dc25e94587128709e819e544b7be9016b255316153c3a8'
+ run: |
+ archive="$RUNNER_TEMP/tofu.zip"
+ curl --fail --location --retry 3 "https://github.com/opentofu/opentofu/releases/download/v${TOFU_VERSION}/tofu_${TOFU_VERSION}_linux_amd64.zip" --output "$archive"
+ printf '%s %s\n' "$TOFU_SHA256" "$archive" | sha256sum --check -
+ mkdir -p "$RUNNER_TEMP/tofu-bin"
+ unzip -q "$archive" tofu -d "$RUNNER_TEMP/tofu-bin"
+ echo "$RUNNER_TEMP/tofu-bin" >> "$GITHUB_PATH"
+ - name: Run acceptance tests
+ run: make test-acceptance-tofu
diff --git a/Makefile b/Makefile
index 0c75cc5..2a2a48a 100644
--- a/Makefile
+++ b/Makefile
@@ -13,7 +13,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-.PHONY: build check check-license fmt fmt-check fmt-terraform test test-acceptance test-race validate-examples vet
+.PHONY: build check check-docs check-license fmt fmt-check fmt-terraform test test-acceptance test-acceptance-tofu test-integration test-race validate-examples vet
build:
go build ./...
@@ -33,9 +33,21 @@ validate-examples: fmt-terraform
test:
go test ./...
+check-docs:
+ go test ./internal/provider -run '^TestDocumentationSchema$$' -count=1
+
test-acceptance:
TF_ACC=1 go test -v ./internal/provider -run '^TestAcc'
+test-acceptance-tofu:
+ @command -v tofu >/dev/null || (echo "OpenTofu must be installed on PATH" >&2; exit 1)
+ TF_ACC=1 TF_ACC_TERRAFORM_PATH="$$(command -v tofu)" TF_ACC_PROVIDER_HOST=registry.opentofu.org go test -v ./internal/provider -run '^TestAcc'
+
+test-integration:
+ @test -n "$$PAIMON_ACC_URI" || (echo "PAIMON_ACC_URI must name an isolated REST Catalog" >&2; exit 1)
+ @test "$$PAIMON_ACC_ALLOW_MUTATIONS" = "1" || (echo "PAIMON_ACC_ALLOW_MUTATIONS=1 is required for live catalog acceptance" >&2; exit 1)
+ TF_ACC=1 go test -v ./internal/provider -run '^TestAccRealCatalog$$' -count=1
+
test-race:
go test -race ./...
diff --git a/README.md b/README.md
index 9da5150..24f8d46 100644
--- a/README.md
+++ b/README.md
@@ -25,8 +25,9 @@ Catalog. It follows the provider shape established by
but talks directly to Paimon's language-neutral REST Catalog API instead of
embedding a JVM client.
-The provider is published from signed Apache release candidates. See the
-[release guide](docs/releasing.md) for the source-vote and Registry workflow.
+Releases use signed Apache release candidates. See the
+[release guide](docs/releasing.md) for the source-vote and Registry workflow,
+including verification of a published version before installation.
## Supported objects
@@ -91,10 +92,10 @@ resource "paimon_table" "events" {
}
]
- primary_keys = ["event_id"]
partition_keys = []
options = {
- "bucket" = "4"
+ "primary-key" = "event_id"
+ "bucket" = "4"
}
comment = "Events managed by Terraform"
}
@@ -118,6 +119,16 @@ Permission, row-filter, and column-mask resources use Paimon's experimental
REST management API. Principal lifecycle and group or role membership remain
server responsibilities.
+Table replacements are blocked unless `allow_replacement = true`.
+Protect retained production data with Terraform's `prevent_destroy` lifecycle
+rule as shown in the [production example](examples/production/main.tf).
+Row-filter and column-mask content updates require
+`allow_non_atomic_update = true` during a maintenance window because the REST
+API cannot update a policy atomically. Import existing policies explicitly.
+
+The [production validation guide](docs/production-readiness.md) distinguishes
+local protocol coverage from real REST/DLF and query-engine validation.
+
## Import
```bash
@@ -129,11 +140,23 @@ terraform import paimon_permission.analyst_read \
## Development
-Go 1.25 or newer is required.
+Use the Go version specified in `go.mod` or a newer compatible version.
```bash
make check
+make check-docs
+make test-acceptance
```
-See [`docs/index.md`](docs/index.md) for the full provider configuration and
-resource notes.
+`make check` includes the documentation schema check. `make check-docs` runs it
+on its own: every registered provider, resource, and data-source field must have
+a reference entry with its actual type, required/optional/computed mode, and
+sensitive flag. Update the reference descriptions when changing defaults,
+validation, or lifecycle behavior; those semantics still require review.
+
+See [`docs/index.md`](docs/index.md) for the provider configuration and links to
+the complete resource and data-source field references.
+
+See the [API contract](docs/api-contract.md) for the current resource model and
+experimental management boundary. Primary keys use `options["primary-key"]`;
+`primary_keys` is a computed output.
diff --git a/dev/release/verify_registry.sh b/dev/release/verify_registry.sh
new file mode 100755
index 0000000..0bbfbbb
--- /dev/null
+++ b/dev/release/verify_registry.sh
@@ -0,0 +1,72 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+set -euo pipefail
+
+version="${1:-}"
+cli="${2:-terraform}"
+if [[ $# -gt 2 || ! "$version" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]; then
+ echo 'Usage: verify_registry.sh VERSION [terraform|tofu|/path/to/cli]' >&2
+ exit 1
+fi
+command -v "$cli" >/dev/null
+command -v python3 >/dev/null
+registry=registry.terraform.io
+if [[ "$(basename "$cli")" = tofu ]]; then
+ registry=registry.opentofu.org
+fi
+
+work_dir="$(mktemp -d -t paimon-registry-check.XXXXXX)"
+trap 'rm -rf "$work_dir"' EXIT
+cat >"$work_dir/main.tf" <<EOF
+terraform {
+ required_providers {
+ paimon = {
+ source = "$registry/apache/paimon"
+ version = "= $version"
+ }
+ }
+}
+EOF
+cat >"$work_dir/cli.tfrc" <<'EOF'
+provider_installation {
+ direct {}
+}
+EOF
+
+# Ignore local mirrors, development overrides, and cached provider installs.
+export TF_CLI_CONFIG_FILE="$work_dir/cli.tfrc"
+export TF_DATA_DIR="$work_dir/data"
+unset TF_PLUGIN_CACHE_DIR TF_PLUGIN_CACHE_MAY_BREAK_DEPENDENCY_LOCK_FILE
+unset TF_CLI_ARGS TF_CLI_ARGS_init TF_CLI_ARGS_validate TF_CLI_ARGS_providers
+"$cli" version
+"$cli" -chdir="$work_dir" init -backend=false -input=false -no-color
+"$cli" -chdir="$work_dir" validate -no-color
+"$cli" -chdir="$work_dir" providers schema -json >"$work_dir/schema.json"
+python3 - "$work_dir/schema.json" "$registry/apache/paimon" <<'PY'
+import json
+import sys
+
+with open(sys.argv[1], encoding="utf-8") as source:
+ schema = json.load(source)["provider_schemas"][sys.argv[2]]
+for name in ("database", "table", "permission", "row_filter", "column_mask"):
+ if "paimon_" + name not in schema.get("resource_schemas", {}):
+ raise SystemExit("Missing resource schema: " + name)
+for name in ("database", "table"):
+ if "paimon_" + name not in schema.get("data_source_schemas", {}):
+ raise SystemExit("Missing data source schema: " + name)
+print("Registry installation and all resource/data-source schemas verified")
+PY
diff --git a/docs/api-contract.md b/docs/api-contract.md
new file mode 100644
index 0000000..a3c0533
--- /dev/null
+++ b/docs/api-contract.md
@@ -0,0 +1,84 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# API contract
+
+The core Terraform API consists of `paimon_database`, `paimon_table`, and their
+single-object data sources. Keep the ordered `fields` list, string-valued
+`options`, explicit `partition_keys`, and computed server metadata. Primary
+keys are configured through Paimon's Java `primary-key` option; `primary_keys`
+is the normalized read-only result. Terraform `id` identifies the managed
+object; `server_id` is the database/table UUID returned by the server, not a
+Catalog identifier. `options` manages declared keys and preserves unmanaged
+keys; `server_options` exposes the raw server map.
+
+## Terraform field references
+
+The following pages cover all registered Terraform inputs and outputs, including
+nested field attributes, types, required/optional/computed modes, sensitive
+flags, defaults, and lifecycle behavior:
+
+- [Provider configuration](index.md)
+- [Database resource](resources/database.md) and [data source](data-sources/database.md)
+- [Table resource](resources/table.md) and [data source](data-sources/table.md)
+- [Permission resource](resources/permission.md)
+- [Row-filter resource](resources/row_filter.md)
+- [Column-mask resource](resources/column_mask.md)
+
+These references describe the provider's Terraform API. Arbitrary Paimon option
+names and the experimental Predicate/Transform JSON vocabulary remain server
+contracts; the provider documents the options it handles specially and delegates
+other validation to the deployed server.
+
+`make check-docs` compares these pages with the schema served through the
+Terraform protocol. It fails for missing or stale field entries, incorrect
+types, modes, or sensitive flags, and empty field descriptions. It also covers
+newly registered resources and data sources. The check runs in the normal Go
+test suite and `make check`; defaults, validators, and behavior descriptions
+must additionally be reviewed against the implementation.
+
+## Reference contracts
+
+The review used these exact source revisions. They are source compatibility
+references, not a minimum version or certification for deployed servers.
+
+| Reference | Contract used |
+| --- | --- |
+| [Paimon 2.0 final, 604e6d5](https://github.com/apache/paimon/tree/604e6d5e131c74a8d127333a2a3ad6d0319732bf) | Schema primary-key normalization, field identity, SchemaChange and constant default conversion |
+| [Paimon main, 475be56](https://github.com/apache/paimon/tree/475be566fef490ad147deec4ee15344c25c0352d) | The same core contract plus experimental permission/policy models and reference REST server policy canonicalization |
+| [terraform-provider-iceberg, c7d15b4](https://github.com/apache/terraform-provider-iceberg/tree/c7d15b42799dd116ba51cf6ed9c43997f41ad2b2) | Comparison of Terraform resource/data-source boundaries, schema evolution and ownership; Iceberg-specific partition transforms are not Paimon partition keys |
+
+Core names and ownership behavior can be maintained as the stable API baseline.
+Future list data sources, nested schema evolution and additional Paimon objects
+can be additive. Table identity changes currently require replacement;
+`allow_replacement = true` explicitly permits plans that can delete data.
+This setting does not protect explicit destroy or resource removal. Supported
+in-place changes remain governed by the deployed Catalog's validation.
+
+`paimon_permission`, `paimon_row_filter`, and `paimon_column_mask` remain
+experimental because the Java management APIs and serialized Predicate/Transform
+AST are experimental. Reference policy canonicalization is implemented in the
+Java test REST server; it is not proof that every external REST/DLF deployment
+behaves identically. Pin the provider and server revisions, run the real catalog
+suite, and validate query enforcement before accepting a deployment. Content
+changes to filters and masks remain non-atomic and require a maintenance window
+with `allow_non_atomic_update = true`.
+
+See [production validation](production-readiness.md) for CLI, deployed-service,
+query-engine and signed Registry checks that source review cannot establish.
diff --git a/docs/data-sources/database.md b/docs/data-sources/database.md
index c791504..49f9684 100644
--- a/docs/data-sources/database.md
+++ b/docs/data-sources/database.md
@@ -32,5 +32,42 @@ data "paimon_database" "analytics" {
}
```
-The data source exports the database catalog ID, location, options, owner, and
-audit timestamps/principals.
+Reads metadata without creating, updating, or importing the remote database.
+A missing database produces an error rather than an empty result. Required lookup
+attributes cannot be omitted or set to `null`.
+
+```hcl
+output "database_server_id" {
+ value = data.paimon_database.analytics.server_id
+}
+```
+
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `name` | `string` | Required | No | Name of the existing database to read. |
+| `id` | `string` | Computed | No | Terraform data-source identity, equal to the database name. |
+| `server_id` | `string` | Computed | No | Server-assigned database object identifier, not the Catalog ID. |
+| `options` | `map(string)` | Computed | No | Complete raw database option map returned by the server. This data source takes no ownership of options. |
+| `location` | `string` | Computed | No | Database location returned by the server. |
+| `owner` | `string` | Computed | No | Owner metadata returned by the server; independent of a similarly named option. |
+| `created_at` | `number` | Computed | No | Creation timestamp in milliseconds since the Unix epoch. |
+| `created_by` | `string` | Computed | No | Principal that created the object, as reported by the server. |
+| `updated_at` | `number` | Computed | No | Last update timestamp in milliseconds since the Unix epoch. |
+| `updated_by` | `string` | Computed | No | Principal that last updated the object, as reported by the server. |
+
+<!-- schema:end -->
+
+## Missing server metadata
+
+Omitted string metadata is reported as `""` and omitted numeric metadata as `0`.
+Empty collections remain empty lists/maps. Do not interpret a zero audit
+timestamp as evidence that the object was created at the Unix epoch.
diff --git a/docs/data-sources/table.md b/docs/data-sources/table.md
index 78ee2eb..519fa18 100644
--- a/docs/data-sources/table.md
+++ b/docs/data-sources/table.md
@@ -33,5 +33,63 @@ data "paimon_table" "events" {
}
```
-The data source exports fields, partition keys, primary keys, options, comment,
-schema ID, path, external-table flag, owner, and audit metadata.
+Reads metadata without creating, updating, or importing the remote table.
+A missing table produces an error rather than an empty result. Required lookup
+attributes cannot be omitted or set to `null`.
+
+```hcl
+output "table_server_id" {
+ value = data.paimon_table.events.server_id
+}
+```
+
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `database` | `string` | Required | No | Database containing the existing table to read. |
+| `name` | `string` | Required | No | Name of the existing table to read. |
+| `id` | `string` | Computed | No | Terraform data-source identity in percent-encoded URL-query form, for example `database=analytics&table=events`. |
+| `server_id` | `string` | Computed | No | Server-assigned table object identifier, distinct from Terraform `id` and the Catalog ID. |
+| `fields` | `list(object)` | Computed | No | All top-level fields in schema order. Nested object attributes are documented below. |
+| `partition_keys` | `list(string)` | Computed | No | Ordered partition field names. Empty for an unpartitioned table. |
+| `primary_keys` | `list(string)` | Computed | No | Normalized, ordered primary-key field names. Empty when the table has no primary keys. |
+| `options` | `map(string)` | Computed | No | Complete raw table option map. Normally excludes the Java-consumed `primary-key` option; read `primary_keys` instead. This data source takes no ownership of options. |
+| `comment` | `string` | Computed | No | Table comment; `null` when absent. An explicit empty string is distinct from absence. |
+| `schema_id` | `number` | Computed | No | Current schema version identifier returned by the server. |
+| `path` | `string` | Computed | No | Table storage path returned by the server. |
+| `is_external` | `bool` | Computed | No | Whether the server reports an external table. |
+| `owner` | `string` | Computed | No | Owner metadata returned by the server; independent of a similarly named option. |
+| `created_at` | `number` | Computed | No | Creation timestamp in milliseconds since the Unix epoch. |
+| `created_by` | `string` | Computed | No | Principal that created the object, as reported by the server. |
+| `updated_at` | `number` | Computed | No | Last update timestamp in milliseconds since the Unix epoch. |
+| `updated_by` | `string` | Computed | No | Principal that last updated the object, as reported by the server. |
+
+### Nested `fields` attributes
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `fields[].id` | `number` | Computed | No | Stable top-level field identifier assigned by Paimon. |
+| `fields[].name` | `string` | Computed | No | Top-level field name. |
+| `fields[].type` | `string` | Computed | No | Paimon SQL type string. Top-level nullability is reported separately in `nullable`. |
+| `fields[].nullable` | `bool` | Computed | No | Whether this field accepts null values. |
+| `fields[].description` | `string` | Computed | No | Field comment; `null` when absent. An empty string is distinct from absence. |
+| `fields[].default_value` | `string` | Computed | No | Constant default represented as a string, or `null` when absent. An empty string and the literal text `NULL` are distinct from no default. |
+| `fields[].nested_field_ids` | `map(number)` | Computed | No | Stable nested ROW field IDs keyed by escaped field paths. Path components escape `~` as `~0` and `/` as `~1`; empty when there are no nested ROW fields. |
+
+<!-- schema:end -->
+
+## Missing server metadata
+
+Comments and field defaults that are absent on the server are reported as `null`
+where those attributes exist. Other omitted string metadata is reported as `""`,
+omitted numeric metadata as `0`, and an omitted external-table flag as `false`.
+Empty collections remain empty lists/maps. Do not interpret a zero audit
+timestamp as evidence that the object was created at the Unix epoch.
diff --git a/docs/index.md b/docs/index.md
index c8fa9d8..cc98d1e 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -40,36 +40,51 @@ provider "paimon" {
}
```
-### Arguments
-
-- `uri` (required): REST Catalog base URI. A base path is supported.
-- `warehouse` (optional): sent as the `warehouse` query parameter to
- `/v1/config`.
-- `token_provider` (optional): `bear` for Bearer authentication or `dlf` for
- Alibaba Cloud DLF AK/STS signing. It is inferred when omitted.
-- `token` (optional, sensitive): token used by the `bear` provider.
-- `prefix` (optional): client catalog path prefix. A server override returned
- by `/v1/config` takes precedence.
-- `headers` (optional, sensitive): additional REST request headers. The
- provider-managed `Authorization` header takes precedence.
-- `dlf_region` (optional): region for DLF default signing. Standard DLF
- endpoints allow it to be inferred.
-- `dlf_signing_algorithm` (optional): `default` for DLF VPC/default endpoints
- or `openapi` for DLFNext endpoints. It is inferred from standard endpoints.
-- `dlf_access_key_id` and `dlf_access_key_secret` (optional, sensitive): a
- static Alibaba Cloud access key pair.
-- `dlf_security_token` (optional, sensitive): STS token paired with static
- access keys.
-- `dlf_token_loader` (optional): `local_file` for a rotating token file or
- `ecs` for ECS RAM role credentials.
-- `dlf_token_path` (optional): path to the rotating AK/STS JSON file. Setting
- it implies the `local_file` loader.
-- `dlf_ecs_metadata_url` (optional): compatible ECS metadata endpoint override.
-- `dlf_ecs_role_name` (optional): RAM role name. The `ecs` loader discovers it
- when omitted.
-
-Exactly one DLF credential source may be configured: static AK/STS, a local
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `uri` | `string` | Required | No | REST Catalog base URI, including an optional base path. Must be absolute HTTP or HTTPS with a host and no query or fragment. |
+| `warehouse` | `string` | Optional | No | Warehouse identifier sent to `/v1/config`. Required for DLF; otherwise omitted when unset. |
+| `request_timeout_seconds` | `number` | Optional | No | Timeout for each REST or credential HTTP request. Default: `30` seconds. Allowed: `1`–`3600`. |
+| `recovery_timeout_seconds` | `number` | Optional | No | Time budget for read-back reconciliation after a mutation. Default: `5` seconds. Allowed: `1`–`3600`. Does not enable mutation retries. |
+| `token_provider` | `string` | Optional | No | Authentication mode: `bear` or `dlf`. Inferred from DLF configuration or a bearer token when omitted. With neither configured, there is no provider-managed authentication. |
+| `token` | `string` | Optional | Yes | Bearer token. Required and nonempty when `token_provider = "bear"`; incompatible with DLF configuration. |
+| `prefix` | `string` | Optional | No | Catalog path prefix for `/v1/{prefix}/…`. Server defaults apply when omitted; server overrides take precedence over this value. |
+| `headers` | `map(string)` | Optional | Yes | Additional REST request headers; none by default. Provider-managed authentication headers take precedence. |
+| `dlf_region` | `string` | Optional | No | Region for DLF default signing. Inferred from standard DLF endpoints; set it explicitly when default signing cannot infer it. |
+| `dlf_signing_algorithm` | `string` | Optional | No | Signing algorithm: `default` or `openapi`. Inferred from the endpoint when omitted: DLF VPC/default endpoints use `default`, DLFNext endpoints use `openapi`. |
+| `dlf_access_key_id` | `string` | Optional | Yes | Static Alibaba Cloud access key ID. Must be paired with `dlf_access_key_secret`; cannot accompany a dynamic credential loader. |
+| `dlf_access_key_secret` | `string` | Optional | Yes | Static Alibaba Cloud access key secret. Must be paired with `dlf_access_key_id`. |
+| `dlf_security_token` | `string` | Optional | Yes | Optional STS token for the static access key pair. Omit for a long-lived AK pair. |
+| `dlf_token_loader` | `string` | Optional | No | Dynamic credential source: `local_file` or `ecs`. A configured `dlf_token_path` implies `local_file`; otherwise there is no default dynamic loader. |
+| `dlf_token_path` | `string` | Optional | No | Path to a rotating AK/STS JSON file. Required for `local_file`; setting it selects that loader when the loader is omitted. |
+| `dlf_ecs_metadata_url` | `string` | Optional | No | ECS metadata endpoint override for the `ecs` loader. Default: `http://100.100.100.200/latest/meta-data/Ram/security-credentials/`. Must be an absolute HTTP(S) URL and support the session-token API described below. |
+| `dlf_ecs_role_name` | `string` | Optional | No | RAM role used by the `ecs` loader. Discovered from the metadata service when omitted. |
+
+<!-- schema:end -->
+
+## Authentication and request behavior
+
+Values in `headers` must be strings; remove a header key to omit it rather than
+setting its value to `null`.
+
+Bearer and DLF configuration are mutually exclusive. An explicit
+`token_provider = "bear"` rejects any nonempty `dlf_*` setting; a bearer `token`
+cannot accompany DLF settings, even when `token_provider` is inferred. Unknown
+values defer the check until configuration resolves.
+
+Configure exactly one DLF credential source: static AK/STS, a local
token file, or an ECS RAM role. DLF Catalog requests also require `warehouse`.
+`dlf_ecs_metadata_url` and `dlf_ecs_role_name` only take effect with the `ecs`
+loader; omit them when using static credentials or a local file.
See the [static STS](../examples/dlf-sts/main.tf),
[rotating token file](../examples/dlf-token-file/main.tf), and
[ECS role](../examples/dlf-ecs/main.tf) examples.
@@ -79,6 +94,44 @@ and server overrides in that order, and then uses the resulting `prefix` for
catalog operations. Redirects are rejected so authentication headers and DLF
signatures are never reused for a different URL.
+The ECS metadata endpoint must support the IMDS session-token API at
+`/latest/api/token` on the same origin; the provider does not fall back to
+tokenless requests.
+
+The ECS loader obtains and caches an IMDS session token, refreshes it before
+expiry, and reacquires it once on an unauthorized metadata response. This
+supports instances with `HttpTokens=required`. Catalog error diagnostics expose
+HTTP/error codes and, when present, a UUID request ID suitable for support
+correlation; arbitrary response text and echoed authentication values are
+suppressed.
+
+### Token file format and rotation
+
+The `local_file` loader reads this JSON structure. Replace the example values
+with credentials and the expiration supplied by your credential issuer:
+
+```json
+{
+ "AccessKeyId": "replace-with-access-key-id",
+ "AccessKeySecret": "replace-with-access-key-secret",
+ "SecurityToken": "replace-with-sts-token",
+ "Expiration": "2026-12-31T23:59:59Z"
+}
+```
+
+`AccessKeyId` and `AccessKeySecret` must be nonblank strings. `SecurityToken`
+is optional for a long-lived AK pair. `Expiration` is an optional RFC3339
+timestamp, but is needed for automatic rotation: without it, credentials remain
+cached for the current provider instance, and replacing the file does not
+trigger a reload.
+
+For credentials with an expiration, the file and ECS loaders attempt refresh
+on a request when less than one hour remains. This is request-driven, not a
+background file watcher. If refresh fails while the cached credentials remain
+valid, the provider continues using them and tries again on a later request;
+expired credentials are not used. Static AK/STS configuration has no automatic
+refresh; select a dynamic loader when credentials need to rotate during a run.
+
## Resources and data sources
- [Database resource](resources/database.md)
@@ -93,3 +146,9 @@ The permission and policy resources use Paimon's experimental REST management
contract. Principal lifecycle, type resolution, role membership, persistence,
and authorization enforcement are server responsibilities. See the
[permissions example](../examples/permissions/main.tf).
+
+See the [production validation guide](production-readiness.md) for integration
+coverage, compatibility evidence, and rollout checks.
+
+See the [API contract](api-contract.md) for the core resource model and
+experimental management boundary.
diff --git a/docs/production-readiness.md b/docs/production-readiness.md
new file mode 100644
index 0000000..55343c2
--- /dev/null
+++ b/docs/production-readiness.md
@@ -0,0 +1,112 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# Production validation
+
+Use an exact provider version and retain the dependency lock file. A passing
+unit or in-process acceptance suite is evidence of provider behavior, not a
+certification of a deployed Catalog or its query engines.
+
+## Coverage and release evidence
+
+| Layer | Repeatable check | Scope / remaining evidence |
+| --- | --- | --- |
+| Provider and client | `make check` | Unit tests, race detector, build, vet, and example validation |
+| Terraform protocol | `make test-acceptance` | Database/table CRUD, schema evolution, immutable-option import, permission/filter/mask CRUD, import and empty plans against an in-process REST fixture |
+| OpenTofu protocol | `make test-acceptance-tofu` | Same lifecycle suite using OpenTofu's registry address and CLI |
+| Real REST Catalog | `make test-integration` | Supplied isolated endpoint; database/table/data-source lifecycle and schema addition; management lifecycle when a test principal is supplied |
+| DLF authentication | Unit tests plus integration with each deployed signing mode | Stable default/OpenAPI signature vectors, rotating credentials, and required-token ECS metadata are covered locally; service authorization and endpoint compatibility require a live run |
+| Query enforcement | Flink/Spark tests against the deployment | Verify actual reads, filters, masks, schema changes, and permission revocation using deployed engine versions |
+| Distribution | `dev/release/verify_registry.sh VERSION CLI` | Install the signed release through the CLI's public registry, validate configuration, and read provider schemas |
+
+The live suite is skipped when no endpoint is supplied. Do not interpret that
+skip as a passing integration check. Record the exact provider commit/release,
+CLI version, Catalog build (including server implementation), DLF region and
+signing mode, engine versions, job URL, and results for each supported deployment.
+No Paimon/DLF minimum server version is claimed from fixture tests. The
+permission and policy APIs are experimental and require a server implementing
+their persistence and authorization contracts.
+
+Protocol CI pins Terraform 1.13.5 and OpenTofu 1.12.6. Update the pinned version
+and its verification checksum together, then rerun the suite before widening
+the supported CLI matrix.
+
+## Running against a real Catalog
+
+Provision an isolated warehouse/catalog and, for all five resources, an
+existing test principal with no running queries. The runner needs rights to
+create, alter, and delete the generated database and its table, and to manage
+that principal's permissions and policies on this table. The suite prints its
+unique `terraform_acc_*` database name for cleanup if the service becomes
+unavailable during teardown.
+
+Configure authentication through the corresponding sensitive `TF_VAR_paimon_*`
+environment variables (see the provider configuration reference). Supply secrets
+through your CI secret store or local secret manager, without embedding them in
+HCL, commands committed to the repository, or test logs. For example, set
+`TF_VAR_paimon_token` for bearer authentication, or use
+`TF_VAR_paimon_dlf_token_loader=local_file` with
+`TF_VAR_paimon_dlf_token_path` and `TF_VAR_paimon_warehouse` for rotating STS.
+A compatible REST endpoint and ECS role configuration work the same way.
+
+```bash
+export PAIMON_ACC_URI=https://isolated-catalog.example.com
+export PAIMON_ACC_ALLOW_MUTATIONS=1
+export PAIMON_ACC_PRINCIPAL=role:terraform-test
+make test-integration
+```
+
+Omitting `PAIMON_ACC_PRINCIPAL` runs only database/table/data-source coverage.
+The GitHub **Live catalog acceptance** workflow requires a principal to cover
+all resources. Configure its `catalog-integration` environment with the secrets
+and variables named in the workflow, and supply the exact server version when
+dispatching. Run separately for plain REST, DLF default signing, and DLF OpenAPI
+where those combinations are supported. The suite deliberately exercises
+non-atomic policy replacement on its isolated table.
+
+The suite checks metadata behavior; a release owner must additionally write
+representative records with the deployed engine, check them before and after
+schema evolution, assert that a restricted principal sees only permitted rows
+and masked columns, and verify revocation denies reads. Retain these results
+with the integration job; do not use metadata equality as evidence of query
+enforcement.
+
+## Operations and recovery
+
+- Import existing objects before managing them. In particular, a rejected
+ policy create does not establish ownership of a pre-existing policy.
+- Review the plan with default replacement protection enabled. For retained
+ data, also use `lifecycle { prevent_destroy = true }`; removing the resource
+ from configuration removes Terraform's lifecycle protection too.
+- Omit IDs for new columns in an existing table. Retain existing IDs to express
+ renames. Changing immutable options or keys requires a deliberate replacement.
+- Pause affected queries before enabling `allow_non_atomic_update` for a policy
+ content change, or replacing/removing a policy. Restore the flag to false
+ afterwards. Rollback can restore old content but cannot make this API atomic.
+- Set `recovery_timeout_seconds` to cover expected visibility delays. After a
+ timeout or transport error, refresh and inspect a new plan before retrying;
+ the provider does not blindly retry writes. Failed removal verification keeps
+ the previous managed keys in state until a read observes the actual result.
+- Keep state and backups in an access-controlled backend. Correlate catalog
+ errors using the UUID request ID when available; never attach credentials,
+ token-file contents, or metadata credential responses to reports.
+
+A production release needs both deployed-server evidence and successful signed
+Registry installation. The scripts and workflows make those checks repeatable;
+they do not replace Apache release voting or publish a release automatically.
diff --git a/docs/releasing.md b/docs/releasing.md
index 9f3c6f0..765985f 100644
--- a/docs/releasing.md
+++ b/docs/releasing.md
@@ -31,6 +31,7 @@ private key.
- Commit access to `dist.apache.org/repos/dist/dev/paimon` and PMC access for
`dist.apache.org/repos/dist/release/paimon`.
- `git`, `svn`, `gh`, `gpg`, Go, Terraform, and `shasum` on `PATH`.
+- OpenTofu and Python 3 for the public Registry installation check.
- A clean checkout whose `origin` is the Apache repository.
Before cutting a candidate, run:
@@ -41,8 +42,13 @@ Before cutting a candidate, run:
make check
make check-license
make test-acceptance
+make test-acceptance-tofu
```
+Complete the [production validation matrix](production-readiness.md) against
+each supported real Catalog/DLF and query-engine deployment, recording exact
+versions and job results before claiming production compatibility.
+
Commit any refreshed `LICENSE-binary`, `NOTICE-binary`, or `licenses-binary/`
files. They describe the code statically linked into every convenience binary.
@@ -83,3 +89,18 @@ ASF `dist/release`, and creates the final GitHub Release from the already-voted
assets. Nothing is rebuilt. Confirm the release appears in both registries,
record it in the ASF reporter, and send an announcement to the Paimon mailing
list.
+
+After each Registry has indexed the final version, verify installation from a
+fresh directory without development overrides or a local plugin mirror:
+
+```bash
+./dev/release/verify_registry.sh 0.1.0 terraform
+./dev/release/verify_registry.sh 0.1.0 tofu
+```
+
+The script pins the requested version, runs the CLI's normal Registry download
+and checksum/signature verification, validates configuration, and checks all
+five resource and two data-source schemas. It does not contact a Catalog or
+create resources. A missing or unindexed release must fail this check; do not
+substitute a local binary when recording distribution evidence. Retain the
+output and the release signing-key identity with the release verification.
diff --git a/docs/resources/column_mask.md b/docs/resources/column_mask.md
index 1ff7d57..646a008 100644
--- a/docs/resources/column_mask.md
+++ b/docs/resources/column_mask.md
@@ -44,6 +44,29 @@ resource "paimon_column_mask" "analyst_email" {
}
```
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `database` | `string` | Required | No | Nonblank database name. Changing it replaces the policy attachment. |
+| `table` | `string` | Required | No | Nonblank table name. The table must exist with `query-auth.enabled=true`. Changing it replaces the policy attachment. |
+| `principal` | `string` | Required | No | Existing server principal identifier. Must be nonblank and at most 128 UTF-16 code units. Changing it replaces the policy attachment. |
+| `column` | `string` | Required | No | Nonblank top-level column name to mask. Changing it replaces the policy attachment. |
+| `transform` | `string` | Required | No | Nonempty JSON serialization of one Paimon `Transform`, at most 60 KiB in UTF-8. The server validates the AST. Meaningful changes require `allow_non_atomic_update = true` and use drop/create. |
+| `allow_non_atomic_update` | `bool` | Optional + computed | No | Default: `false`. Permits content updates by dropping and recreating the policy. Does not guard explicit deletion or identity replacement; see the maintenance requirements below. |
+| `id` | `string` | Computed | No | Percent-encoded URL-query identity containing database, table, principal, and column. Use the same identity for import. |
+
+<!-- schema:end -->
+
+## Behavior
+
This example replaces every email value with the constant `***@***.com`; it
does not include the original column value in the transform output.
@@ -54,10 +77,26 @@ JSON formatting and object-key ordering alone do not replace the remote policy;
an in-place apply only updates the stored representation.
Paimon currently has create and drop operations but no policy update operation.
-Changing `transform` therefore drops and recreates the policy. If creation
-fails, the provider attempts to restore the previous policy and reports whether
-restoration succeeded. There can still be a brief interval without the mask;
-schedule sensitive changes accordingly.
+Changing `transform` is blocked by default. Set
+`allow_non_atomic_update = true` only in a maintenance window with affected
+queries paused: the provider drops and recreates the policy. If creation fails,
+it attempts to restore the previous policy and reports whether restoration
+succeeded. This opt-in and rollback do not eliminate the interval without the
+mask. Setting the flag alone or changing only JSON formatting does not
+recreate the policy. The flag controls content updates, not explicit deletion
+or replacement following an attachment/identity change; keep affected queries
+paused for those operations too.
+
+A definitively rejected initial create (including HTTP 409 or 403) does not
+adopt an existing policy, even if its content matches. Import it explicitly to
+establish Terraform ownership. An uncertain response such as a connection loss
+is reconciled by reading the exact policy identity and content.
+
+A refresh that finds the policy missing removes it from Terraform state; a
+subsequent plan can recreate it. Required attributes cannot be omitted or set
+to `null`. Omitted/null `allow_non_atomic_update` resolves to `false`.
+
+## Import
Import with the URL-query identity printed in `id`:
@@ -65,3 +104,12 @@ Import with the URL-query identity printed in `id`:
terraform import paimon_column_mask.analyst_email \
'database=analytics&table=events&principal=role%3Aanalyst&column=email'
```
+
+The provider compares known Java policy AST semantics across create, refresh,
+and update: field references resolve by name against the current table schema,
+and one-child AND/OR predicates simplify to their child. Equivalent configured
+JSON is retained in state. Changes to field names, functions, literal values,
+cast target types, or unknown AST members remain meaningful differences. The
+Catalog must permit reading the protected table schema for this comparison.
+See the [API contract](../api-contract.md) for the exact Java
+reference revision and experimental compatibility boundary.
diff --git a/docs/resources/database.md b/docs/resources/database.md
index 7de732c..13934be 100644
--- a/docs/resources/database.md
+++ b/docs/resources/database.md
@@ -37,13 +37,46 @@ resource "paimon_database" "example" {
}
```
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `name` | `string` | Required | No | Database name. Changing it replaces the resource. |
+| `options` | `map(string)` | Optional | No | Options managed by Terraform. Initially omitted/null means no managed keys. Removing previously managed keys, including omitting the whole map, deletes those keys in place; unrelated server keys are preserved. |
+| `id` | `string` | Computed | No | Terraform identity, equal to the database name; also used for import. |
+| `server_id` | `string` | Computed | No | Server-assigned database identifier. This is the database object ID, not the Catalog ID. |
+| `server_options` | `map(string)` | Computed | No | Complete raw option map returned by the server, including unmanaged keys. |
+| `location` | `string` | Computed | No | Database location returned by the server. |
+| `owner` | `string` | Computed | No | Owner metadata returned by the server; independent of a similarly named option. |
+| `created_at` | `number` | Computed | No | Creation timestamp in milliseconds since the Unix epoch. |
+| `created_by` | `string` | Computed | No | Principal that created the object, as reported by the server. |
+| `updated_at` | `number` | Computed | No | Last update timestamp in milliseconds since the Unix epoch. |
+| `updated_by` | `string` | Computed | No | Principal that last updated the object, as reported by the server. |
+
+<!-- schema:end -->
+
+## Behavior
+
`name` is required and changing it replaces the resource. `options` contains
only values managed by Terraform; `server_options` exposes the complete map
returned by the catalog. Updates remove only keys previously managed by this
resource, preserving unrelated server values.
+Map entries must contain strings; remove a key rather than setting its value
+to `null`.
+
+After import, omitted `options` leaves remote options unmanaged. Declaring an
+option with its current server value takes ownership without changing that
+value. A refresh that finds the database missing removes it from Terraform
+state; a subsequent plan can recreate it.
-The resource also exports `id`, `catalog_id`, `location`, `owner`,
-`created_at`, `created_by`, `updated_at`, and `updated_by`.
+## Import
Import with the database name:
diff --git a/docs/resources/permission.md b/docs/resources/permission.md
index f483526..8f7b576 100644
--- a/docs/resources/permission.md
+++ b/docs/resources/permission.md
@@ -41,11 +41,45 @@ resource "paimon_permission" "read_events" {
}
```
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `resource_type` | `string` | Required | No | Exact uppercase resource type from the locator table below. Determines required locators and allowed accesses. Changing it replaces the assignment. |
+| `access` | `string` | Required | No | Access from the supported-access table below; must use exact uppercase spelling. Changing it replaces the assignment. |
+| `principal` | `string` | Required | No | Existing server principal identifier. Must be nonblank and at most 128 UTF-16 code units. Changing it replaces the assignment. |
+| `database` | `string` | Optional | No | Required, nonblank database locator for database, table, column, function, and view types; forbidden for catalog types. Changing it replaces the assignment. |
+| `table` | `string` | Optional | No | Required, nonblank table locator for `TABLE` and `COLUMN`; forbidden for other types. Changing it replaces the assignment. |
+| `function` | `string` | Optional | No | Required, nonblank function locator for `FUNCTION`; forbidden for other types. Changing it replaces the assignment. |
+| `view` | `string` | Optional | No | Required, nonblank view locator for `VIEW`; forbidden for other types. Changing it replaces the assignment. |
+| `column_names` | `set(string)` | Optional | No | Allowed columns for a `COLUMN` assignment. Exactly one nonempty column range is required; names must be nonblank and non-null. Forbidden for other types. Changes use grant-or-replace. |
+| `excluded_column_names` | `set(string)` | Optional | No | Excluded columns for a `COLUMN` assignment, mutually exclusive with `column_names`. Same validation and update behavior; newly added columns remain allowed. |
+| `expire_time` | `string` | Optional | No | Exclusive expiry instant in UTC, as detailed below. Omitted/null means no expiry and removes a previously managed expiry; an empty string is invalid. Changes use grant-or-replace. |
+| `id` | `string` | Computed | No | Percent-encoded URL-query identity containing resource type, locators, access, and principal. Expiry and column ranges are content, not part of the identity. |
+
+<!-- schema:end -->
+
+## Behavior
+
The assignment identity is `resource_type` plus its locator, `access`, and
`principal`. Changing an identity attribute replaces the resource. Changing
`expire_time`, `column_names`, or `excluded_column_names` uses Paimon's
grant-or-replace operation.
+Optional locators must be omitted or `null` when not applicable; unrelated
+locators are rejected. Column ranges are sets, so order does not matter. Empty
+sets do not satisfy the required range for a `COLUMN` assignment. Deletion
+revokes this direct assignment; principal lifecycle and inherited access remain
+server responsibilities. A refresh that finds the assignment missing removes
+it from Terraform state.
+
Resource locators have these exact shapes:
| `resource_type` | Locator attributes |
@@ -92,6 +126,8 @@ Equivalent spellings are normalized to upper-case `T`/`Z` with no more than
three fractional digits before they are sent to the server. Expiry is an
exclusive upper bound evaluated by the server clock.
+## Import
+
Import with the URL-query identity printed in `id`:
```bash
diff --git a/docs/resources/row_filter.md b/docs/resources/row_filter.md
index 4aeaeef..009dafd 100644
--- a/docs/resources/row_filter.md
+++ b/docs/resources/row_filter.md
@@ -52,6 +52,28 @@ resource "paimon_row_filter" "analyst_region" {
}
```
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `database` | `string` | Required | No | Nonblank database name. Changing it replaces the policy attachment. |
+| `table` | `string` | Required | No | Nonblank table name. The table must exist with `query-auth.enabled=true`. Changing it replaces the policy attachment. |
+| `principal` | `string` | Required | No | Existing server principal identifier. Must be nonblank and at most 128 UTF-16 code units. Changing it replaces the policy attachment. |
+| `predicate` | `string` | Required | No | Nonempty JSON serialization of one Paimon `Predicate`, at most 60 KiB in UTF-8. The server validates the AST. Meaningful changes require `allow_non_atomic_update = true` and use drop/create. |
+| `allow_non_atomic_update` | `bool` | Optional + computed | No | Default: `false`. Permits content updates by dropping and recreating the policy. Does not guard explicit deletion or identity replacement; see the maintenance requirements below. |
+| `id` | `string` | Computed | No | Percent-encoded URL-query identity containing database, table, principal. Use the same identity for import. |
+
+<!-- schema:end -->
+
+## Behavior
+
`predicate` must be the JSON serialization of one Paimon `Predicate` and must
not exceed 60 KiB in UTF-8. The server validates field references and returns a
canonical representation. JSON formatting and object-key ordering alone do not
@@ -59,10 +81,26 @@ replace the remote policy; an in-place apply only updates the stored
representation.
Paimon currently has create and drop operations but no policy update operation.
-Changing `predicate` therefore drops and recreates the policy. If creation
-fails, the provider attempts to restore the previous policy and reports whether
-restoration succeeded. There can still be a brief interval without the filter;
-schedule sensitive changes accordingly.
+Changing `predicate` is blocked by default. Set
+`allow_non_atomic_update = true` only in a maintenance window with affected
+queries paused: the provider drops and recreates the policy. If creation fails,
+it attempts to restore the previous policy and reports whether restoration
+succeeded. This opt-in and rollback do not eliminate the interval without the
+filter. Setting the flag alone or changing only JSON formatting does not
+recreate the policy. The flag controls content updates, not explicit deletion
+or replacement following an attachment/identity change; keep affected queries
+paused for those operations too.
+
+A definitively rejected initial create (including HTTP 409 or 403) does not
+adopt an existing policy, even if its content matches. Import it explicitly to
+establish Terraform ownership. An uncertain response such as a connection loss
+is reconciled by reading the exact policy identity and content.
+
+A refresh that finds the policy missing removes it from Terraform state; a
+subsequent plan can recreate it. Required attributes cannot be omitted or set
+to `null`. Omitted/null `allow_non_atomic_update` resolves to `false`.
+
+## Import
Import with the URL-query identity printed in `id`:
@@ -70,3 +108,12 @@ Import with the URL-query identity printed in `id`:
terraform import paimon_row_filter.analyst_region \
'database=analytics&table=events&principal=role%3Aanalyst'
```
+
+The provider compares known Java policy AST semantics across create, refresh,
+and update: field references resolve by name against the current table schema,
+and one-child AND/OR predicates simplify to their child. Equivalent configured
+JSON is retained in state. Changes to field names, functions, literal values,
+cast target types, or unknown AST members remain meaningful differences. The
+Catalog must permit reading the protected table schema for this comparison.
+See the [API contract](../api-contract.md) for the exact Java
+reference revision and experimental compatibility boundary.
diff --git a/docs/resources/table.md b/docs/resources/table.md
index 019a96f..6087e88 100644
--- a/docs/resources/table.md
+++ b/docs/resources/table.md
@@ -45,19 +45,81 @@ resource "paimon_table" "example" {
}
]
- primary_keys = ["id"]
options = {
- bucket = "4"
+ "primary-key" = "id"
+ bucket = "4"
}
}
```
+## Schema
+
+Required attributes must be configured. Optional attributes may be omitted or
+set to `null`; their behavior is described below. Optional + computed attributes
+can be configured or resolved by the provider. Computed attributes are read-only.
+`number` attributes and values of `map(number)` are integers.
+
+<!-- schema:begin -->
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `database` | `string` | Required | No | Database containing the table. Changing it requires replacement and `allow_replacement = true`. |
+| `name` | `string` | Required | No | Table name. Changing it requires replacement and `allow_replacement = true`. |
+| `fields` | `list(object)` | Required | No | Ordered, complete list of top-level fields. Removing an entry requests a column drop. See the nested schema and supported evolution below. |
+| `partition_keys` | `list(string)` | Optional + computed | No | Ordered partition field names. Omitted/null means no partitions on creation and inherits existing keys on update/import. Explicit `[]` requests no partitions. Changing keys requires replacement. |
+| `options` | `map(string)` | Optional | No | String-valued Paimon options managed by Terraform, including `primary-key`. No managed keys initially when omitted/null. Removing managed keys deletes them; other server keys survive. Mutable changes update in place; immutable changes require replacement. `partition` is reserved. |
+| `comment` | `string` | Optional | No | Table comment. Omitted/null means no comment and clears a previous comment, including after import. `""` is an explicit empty string. Changes update in place. |
+| `allow_replacement` | `bool` | Optional + computed | No | Default: `false`. Set to `true` to permit required replacement, which can delete table data. Does not block explicit destroy or resource removal. |
+| `id` | `string` | Computed | No | Terraform/import identity in URL-query form, for example `database=analytics&table=events`, with names percent-encoded. |
+| `server_id` | `string` | Computed | No | Server-assigned table object identifier; distinct from Terraform `id` and the Catalog ID. |
+| `primary_keys` | `list(string)` | Computed | No | Normalized, ordered primary-key field names. Configure keys with `options["primary-key"]`, not this attribute. |
+| `server_options` | `map(string)` | Computed | No | Complete raw server option map. Normally excludes the `primary-key` option consumed by Java schema normalization; use `primary_keys` to read those keys. |
+| `schema_id` | `number` | Computed | No | Current schema version identifier returned by the server. |
+| `path` | `string` | Computed | No | Table storage path returned by the server. |
+| `is_external` | `bool` | Computed | No | Whether the server reports an external table. This is an output, not an external-table creation setting. |
+| `owner` | `string` | Computed | No | Owner metadata returned by the server; independent of a similarly named option. |
+| `created_at` | `number` | Computed | No | Creation timestamp in milliseconds since the Unix epoch. |
+| `created_by` | `string` | Computed | No | Principal that created the object, as reported by the server. |
+| `updated_at` | `number` | Computed | No | Last update timestamp in milliseconds since the Unix epoch. |
+| `updated_by` | `string` | Computed | No | Principal that last updated the object, as reported by the server. |
+
+### Nested `fields` attributes
+
+| Attribute | Type | Mode | Sensitive | Meaning, default, and update behavior |
+| --- | --- | --- | --- | --- |
+| `fields[].name` | `string` | Required | No | Unique top-level field name. Retain its ID when renaming an existing field. |
+| `fields[].type` | `string` | Required | No | Nonempty Paimon SQL type string, such as `BIGINT`, `STRING`, or `ROW<item STRING>`. A `NOT NULL` suffix must agree with `nullable = false`. |
+| `fields[].id` | `number` | Optional + computed | No | Stable, unique integer ID from `0` through `1073741822`. May be supplied on initial creation or to identify a retained field. Omit for a new field added to an existing table so Paimon assigns it. |
+| `fields[].nullable` | `bool` | Optional + computed | No | Omitted/null inherits a retained field’s nullability. New non-key fields default to `true` unless the type has `NOT NULL`; primary keys follow `primary-key.nullable`, default `false`. |
+| `fields[].description` | `string` | Optional | No | Field comment. Omitted/null clears an existing description, including after import. `""` is an explicit empty string. Updates in place. |
+| `fields[].default_value` | `string` | Optional | No | Constant string cast by Paimon to the field type, not a SQL expression. Omitted/null clears an existing default; `""` is a distinct constant. Updates in place when supported by the server. |
+| `fields[].nested_field_ids` | `map(number)` | Computed | No | Stable IDs of nested ROW fields keyed by escaped field paths. Read-only; retained across supported changes. In path components, `~0` encodes `~` and `~1` encodes `/`. |
+
+<!-- schema:end -->
+
+## Behavior
+
Each field supports `id`, `name`, `type`, `nullable`, `description`, and
`default_value`. `nested_field_ids` is a computed map that preserves the stable
IDs of nested ROW fields. Top-level field IDs must be unique integers from 0
-through 1073741822; the next available ID is assigned when omitted. Use
+through 1073741822. IDs may be specified when creating a table or retaining an
+existing field (including a rename). Omit `id` when adding a field to an
+existing table: Paimon assigns that ID, and the provider reads it back. Explicit
+new IDs are rejected before mutation. Use
canonical Paimon SQL type strings such as `INT`, `BIGINT`, `STRING`,
`DECIMAL(12, 2)`, `ARRAY<STRING>`, or `ROW<item STRING>`.
+An explicit `NOT NULL` suffix is preserved across reads and must agree with
+`nullable = false`; spelling differences alone do not cause perpetual plans.
+
+`default_value` is a constant represented as a string, matching Java
+`DefaultValueUtils`, not an evaluated SQL expression. Examples include `"42"`
+for BIGINT, `"true"` for BOOLEAN, and `"2026-01-02"` for DATE. The server casts
+the string to the field type and validates supported conversions. `"1 + 2"`
+does not compute a BIGINT and `"CURRENT_TIMESTAMP"` does not compute a timestamp;
+for STRING they are literal text. Omit the attribute or set it to HCL `null`
+to remove a default. `""` is an explicit empty string, distinct from no default;
+`"NULL"` is text, not HCL null. The provider passes constants unchanged and does
+not attempt to reproduce the server's cast engine.
Supported top-level field additions, drops, renames, atomic type/nullability
changes, comments, defaults, and position changes use Paimon's in-place
@@ -65,17 +127,105 @@ changes, comments, defaults, and position changes use Paimon's in-place
Rename cycles require an intermediate name. Adding a non-nullable field,
changing the type of a retained partition/primary-key field, or changing an
existing composite `ROW`, `ARRAY`, `MAP`, `MULTISET`, or `VECTOR` shape replaces
-the table because Paimon's schema-evolution API cannot safely apply those
-changes in place. `database`, `name`, `partition_keys`, and `primary_keys` also
-remain replacement attributes.
-Configure keys with `primary_keys` and `partition_keys`; the
-normalized Paimon options `primary-key` and `partition` are rejected in
-`options` because the server removes them from its options map.
+the table because the provider does not yet implement those changes in place.
+Paimon itself supports some nested field evolution; this is a provider limitation.
+`database`, `name`, `partition_keys`, and `options["primary-key"]` also require
+replacement. These changes are rejected by default. Set
+`allow_replacement = true` only when you intend to replace the table
+and have accounted for data deletion. This setting controls replacement plans;
+it does not block an explicit destroy or removal of the resource from
+configuration.
+On an existing table, configured `database`, `name`, or `partition_keys` values
+that are unknown during planning also require replacement opt-in: the provider
+cannot establish that they are unchanged until the dependency resolves.
+Configure primary keys using the Java option `options["primary-key"] = "id,tenant"`.
+Paimon trims each comma-separated name and ignores empty entries; order remains
+significant. `primary_keys` is a computed output, not a configuration argument.
+The provider sends the option at creation, reads normalized schema keys back,
+and preserves equivalent option spelling in state. `server_options` contains
+the raw server map, which normally omits the consumed `primary-key` option.
+Configure partition keys with `partition_keys`; `options["partition"]` is rejected.
+
+On import, omitted keys and field nullability inherit the existing table.
+Adding the same `primary-key` option later acquires management without an ALTER
+or replacement. Removing a previously managed `primary-key` option means an
+empty primary key list and requires replacement when the old list was nonempty.
+An explicitly empty option likewise requests no primary keys. For new fields,
+omitted `nullable` defaults to true except for primary keys, whose nullability
+follows `primary-key.nullable` (false by default). An explicitly configured
+primary-key field's nullability must agree with that option.
Mutable `options` and `comment` update in place. Changing or removing an option
that Paimon defines as immutable, such as `merge-engine`, `bucket-key`, `type`,
-or `primary-key.nullable`, replaces the table. Unmanaged server options are
-preserved and exposed through `server_options`.
+or `primary-key.nullable`, requires the same replacement opt-in. After import,
+adding an option to configuration with the value already present in
+`server_options` takes ownership without replacing or altering the table.
+Unmanaged server options are preserved and exposed through `server_options`.
+Option removals are successful only when a read confirms their absence. If
+verification times out, the provider reports an error and retains the previous
+managed keys; refresh observes the eventual result and the next plan can retry
+any remaining change.
+
+### Omitted and null values
+
+Omission and HCL `null` have the same meaning for optional attributes, but that
+meaning depends on the attribute. Map entries must contain strings: remove an
+option's key from the map rather than assigning `null` to its value.
+
+| Input | New table | Existing or imported table |
+| --- | --- | --- |
+| `partition_keys` | No partition keys | Inherit existing keys; use `[]` to request removal |
+| `options` | No managed options | Preserve unmanaged keys and remove keys previously managed by this resource |
+| Omitted `primary-key` entry in `options` | No primary keys | Inherit unmanaged keys; removing a previously managed option requests no primary keys |
+| `comment`, `fields[].description`, `fields[].default_value` | Absent | Clear the previous value; these do not inherit imported values |
+| `fields[].id`, `fields[].nullable` | Resolve using the field rules above | Inherit the retained field's identity/nullability |
+| `allow_replacement` | `false` | `false`; opt in explicitly for each configuration that needs replacement |
+
+The `fields` list is always authoritative. Import does not make omitted field
+entries unmanaged. Include every field you intend to retain. Each partition or
+primary-key name must refer to a field, and key lists must not contain duplicates.
+`partition_keys` elements may be unknown during planning, but must not be `null`.
+A refresh that finds the table missing removes it from Terraform state; a
+subsequent plan can recreate it.
+
+### Immutable options
+
+The provider requires replacement when the effective value of any of these
+managed options changes or is removed:
+
+```
+aggregation.remove-record-on-delete
+blob-descriptor-field
+blob-field
+blob-view-field
+bucket-function.type
+bucket-key
+data-evolution.enabled
+data-file.path-directory
+dynamic-bucket.initial-buckets
+force-lookup
+index-file-in-data-file-dir
+merge-engine
+partial-update.remove-record-on-delete
+partial-update.remove-record-on-sequence-group
+pk-clustering-override
+primary-key
+primary-key.nullable
+row-tracking.enabled
+rowkind.field
+sequence.snapshot-ordering
+type
+video-frame-field
+```
+
+`partition` is also immutable in Paimon, but cannot be configured in `options`;
+use `partition_keys`. For `type`, omission has the effective default `table`,
+and comparisons ignore case. Other option names and values are passed through
+to the deployed server, which validates their support and constraints. This
+reference documents the provider's handling of options, not every Paimon engine
+option.
+
+### Destruction
Dropping a managed table can delete its data. Use `prevent_destroy` where
appropriate:
@@ -86,6 +236,8 @@ lifecycle {
}
```
+## Import
+
Import with the unambiguous URL-query identity. The legacy `database.table`
form remains accepted when names do not contain dots:
diff --git a/examples/main.tf b/examples/main.tf
index cc35bb3..fe59e85 100644
--- a/examples/main.tf
+++ b/examples/main.tf
@@ -56,8 +56,10 @@ resource "paimon_table" "events" {
}
]
- primary_keys = ["event_id"]
options = {
+
+ "primary-key" = "event_id"
+
bucket = "4"
}
}
diff --git a/examples/main.tf b/examples/production/main.tf
similarity index 59%
copy from examples/main.tf
copy to examples/production/main.tf
index cc35bb3..029a768 100644
--- a/examples/main.tf
+++ b/examples/production/main.tf
@@ -21,43 +21,53 @@ terraform {
}
}
-variable "paimon_token" {
+variable "catalog_uri" {
+ type = string
+}
+
+variable "warehouse" {
+ type = string
+}
+
+variable "catalog_token" {
type = string
sensitive = true
- default = null
}
provider "paimon" {
- uri = "http://localhost:8080"
- token = var.paimon_token
+ uri = var.catalog_uri
+ warehouse = var.warehouse
+ token_provider = "bear"
+ token = var.catalog_token
+ request_timeout_seconds = 30
+ recovery_timeout_seconds = 60
}
resource "paimon_database" "analytics" {
name = "analytics"
+
+ lifecycle {
+ prevent_destroy = true
+ }
}
resource "paimon_table" "events" {
- database = paimon_database.analytics.name
- name = "events"
-
+ database = paimon_database.analytics.name
+ name = "events"
+ allow_replacement = false
fields = [
- {
- name = "event_id"
- type = "BIGINT"
- nullable = false
- },
- {
- name = "event_time"
- type = "TIMESTAMP(3)"
- },
- {
- name = "payload"
- type = "STRING"
- }
+ { name = "id", type = "BIGINT", nullable = false },
+ { name = "payload", type = "STRING" }
]
- primary_keys = ["event_id"]
options = {
- bucket = "4"
+ "primary-key" = "id"
+
+ "bucket" = "4"
+ "merge-engine" = "deduplicate"
+ }
+
+ lifecycle {
+ prevent_destroy = true
}
}
diff --git a/internal/client/client.go b/internal/client/client.go
index 3ac8341..6442c0a 100644
--- a/internal/client/client.go
+++ b/internal/client/client.go
@@ -26,6 +26,7 @@ import (
"io"
"net/http"
"net/url"
+ "regexp"
"sort"
"strconv"
"strings"
@@ -44,22 +45,25 @@ const (
var errMutationOutcomeUncertain = errors.New("call Paimon REST API")
type Config struct {
- URI string
- Warehouse string
- AuthProvider string
- Token string
- DLF *DLFConfig
- Prefix string
- Headers map[string]string
- HTTPClient *http.Client
+ URI string
+ Warehouse string
+ AuthProvider string
+ Token string
+ DLF *DLFConfig
+ Prefix string
+ Headers map[string]string
+ HTTPClient *http.Client
+ RequestTimeout time.Duration
+ RecoveryTimeout time.Duration
}
type Client struct {
- baseURL *url.URL
- warehouse string
- userPrefix string
- httpClient *http.Client
- auth requestAuthenticator
+ baseURL *url.URL
+ warehouse string
+ userPrefix string
+ httpClient *http.Client
+ auth requestAuthenticator
+ recoveryTimeout time.Duration
mu sync.Mutex
configured bool
@@ -68,19 +72,50 @@ type Client struct {
}
type APIError struct {
- StatusCode int
+ StatusCode int `json:"-"`
Code int `json:"code"`
Message string `json:"message"`
ResourceType *string `json:"resourceType"`
ResourceName *string `json:"resourceName"`
+ RequestID string `json:"-"`
}
func (e *APIError) Error() string {
+ detail := fmt.Sprintf("Paimon REST API returned HTTP %d", e.StatusCode)
if e.Code != 0 {
- return fmt.Sprintf("Paimon REST API returned HTTP %d with code %d", e.StatusCode, e.Code)
+ detail += fmt.Sprintf(" with code %d", e.Code)
+ }
+ if e.RequestID != "" {
+ detail += " (request ID " + e.RequestID + ")"
+ }
+
+ return detail
+}
+
+var requestIDPattern = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`)
+
+// Only expose conventional UUID correlation IDs, never arbitrary response text
+// or an echoed authentication header.
+func safeResponseRequestID(response *http.Response) string {
+ for _, name := range []string{"X-Acs-Request-Id", "X-Dlf-Request-Id", "X-Request-Id"} {
+ candidate := response.Header.Get(name)
+ if !requestIDPattern.MatchString(candidate) {
+ continue
+ }
+ if response.Request != nil {
+ for _, values := range response.Request.Header {
+ for _, value := range values {
+ if strings.Contains(value, candidate) {
+ return ""
+ }
+ }
+ }
+ }
+
+ return candidate
}
- return fmt.Sprintf("Paimon REST API returned HTTP %d", e.StatusCode)
+ return ""
}
func IsNotFound(err error) bool {
@@ -116,6 +151,12 @@ func New(config Config) (*Client, error) {
}
httpClient := noRedirectHTTPClient(config.HTTPClient)
+ if config.RequestTimeout < 0 || config.RecoveryTimeout < 0 {
+ return nil, errors.New("request and recovery timeouts must be positive")
+ }
+ if config.RequestTimeout > 0 {
+ httpClient.Timeout = config.RequestTimeout
+ }
configuredAuth := strings.ToLower(strings.TrimSpace(config.AuthProvider))
if configuredAuth == "" && config.Token != "" {
@@ -151,16 +192,26 @@ func New(config Config) (*Client, error) {
}
return &Client{
- baseURL: baseURL,
- warehouse: config.Warehouse,
- userPrefix: config.Prefix,
- httpClient: httpClient,
- auth: auth,
- prefix: config.Prefix,
- headers: cloneMap(config.Headers),
+ baseURL: baseURL,
+ warehouse: config.Warehouse,
+ userPrefix: config.Prefix,
+ httpClient: httpClient,
+ recoveryTimeout: config.RecoveryTimeout,
+ auth: auth,
+ prefix: config.Prefix,
+ headers: cloneMap(config.Headers),
}, nil
}
+// RecoveryTimeout bounds provider reconciliation separately from each request.
+func (c *Client) RecoveryTimeout() time.Duration {
+ if c == nil || c.recoveryTimeout == 0 {
+ return 5 * time.Second
+ }
+
+ return c.recoveryTimeout
+}
+
func (c *Client) CreateDatabase(ctx context.Context, name string, options map[string]string) error {
request := createDatabaseRequest{Name: name, Options: nonNilMap(options)}
@@ -348,7 +399,7 @@ func (c *Client) doRaw(ctx context.Context, method string, segments []string, qu
defer response.Body.Close()
if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
- apiErr := &APIError{StatusCode: response.StatusCode}
+ apiErr := &APIError{StatusCode: response.StatusCode, RequestID: safeResponseRequestID(response)}
limited := io.LimitReader(response.Body, 1<<20)
if err := json.NewDecoder(limited).Decode(apiErr); err != nil && err != io.EOF {
apiErr.Message = http.StatusText(response.StatusCode)
@@ -357,6 +408,12 @@ func (c *Client) doRaw(ctx context.Context, method string, segments []string, qu
return apiErr
}
+ // Mutations do not consume a response representation. Once the server has
+ // confirmed success, a truncated unused body must not turn it into a failed
+ // create with no Terraform identity. The resource verifies state by reading it.
+ if result == nil || response.StatusCode == http.StatusNoContent {
+ return nil
+ }
contents, err := io.ReadAll(io.LimitReader(response.Body, maxAPIResponseBodySize+1))
if err != nil {
return errors.New("read Paimon REST response")
@@ -365,9 +422,6 @@ func (c *Client) doRaw(ctx context.Context, method string, segments []string, qu
return errors.New("Paimon REST response exceeded 16 MiB size limit")
}
- if result == nil || response.StatusCode == http.StatusNoContent {
- return nil
- }
if err := json.Unmarshal(contents, result); err != nil {
return fmt.Errorf("decode Paimon REST response: %w", err)
}
diff --git a/internal/client/client_test.go b/internal/client/client_test.go
index fed6337..81b9e22 100644
--- a/internal/client/client_test.go
+++ b/internal/client/client_test.go
@@ -20,11 +20,13 @@ package client
import (
"context"
"encoding/json"
+ "io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -329,7 +331,7 @@ func TestSchemaMarshalAssignsUniqueNestedFieldIDs(t *testing.T) {
{
ID: 7,
Name: "payload",
- Type: DataType("ROW<`item name` ARRAY<MAP<STRING NOT NULL, ROW<value VECTOR<DOUBLE, 3>>>> COMMENT 'item''s label' DEFAULT CAST(NULL AS STRING)>"),
+ Type: DataType("ROW<`item name` ARRAY<MAP<STRING NOT NULL, ROW<value VECTOR<DOUBLE, 3>>>> COMMENT 'item''s label'>"),
},
},
}
@@ -345,8 +347,7 @@ func TestSchemaMarshalAssignsUniqueNestedFieldIDs(t *testing.T) {
"id":8,
"name":"item name",
"type":{"type":"ARRAY","element":{"type":"MAP","key":"STRING NOT NULL","value":{"type":"ROW","fields":[{"id":9,"name":"value","type":{"type":"VECTOR","element":"DOUBLE","length":3}}]}}},
- "description":"item's label",
- "defaultValue":"CAST(NULL AS STRING)"
+ "description":"item's label"
}]
}}
],
@@ -428,6 +429,9 @@ func TestDataTypeMarshalSupportsEmptyRow(t *testing.T) {
assert.JSONEq(t, `{"type":"ROW NOT NULL","fields":[]}`, string(encoded))
}
+// Lexical preservation only: these opaque expressions are intentionally not
+// examples of Java default conversion. The Catalog casts constant strings and
+// rejects expressions that cannot be cast to the field's type.
func TestDataTypeMarshalKeepsComparisonsInNestedDefaults(t *testing.T) {
for _, test := range []struct {
name string
@@ -617,3 +621,93 @@ func writeJSON(t *testing.T, w http.ResponseWriter, value any) {
w.Header().Set("Content-Type", "application/json")
require.NoError(t, json.NewEncoder(w).Encode(value))
}
+
+func TestMutationSuccessDoesNotDependOnUnusedResponseBody(t *testing.T) {
+ var created atomic.Bool
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeJSON(t, w, ConfigResponse{})
+
+ return
+ }
+ if r.Method == http.MethodPost {
+ created.Store(true)
+ w.Header().Set("Content-Length", "10")
+ w.WriteHeader(http.StatusOK)
+ _, _ = w.Write([]byte("{}"))
+
+ return
+ }
+ writeJSON(t, w, Database{Name: "analytics"})
+ }))
+ defer server.Close()
+ api, err := New(Config{URI: server.URL})
+ require.NoError(t, err)
+ require.NoError(t, api.CreateDatabase(context.Background(), "analytics", nil))
+ require.True(t, created.Load())
+ database, err := api.GetDatabase(context.Background(), "analytics")
+ require.NoError(t, err)
+ require.Equal(t, "analytics", database.Name)
+}
+
+func TestRequestTimeoutPreservesUncertainMutationOutcome(t *testing.T) {
+ var calls atomic.Int32
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeJSON(t, w, ConfigResponse{})
+
+ return
+ }
+ calls.Add(1)
+ _, _ = io.Copy(io.Discard, r.Body)
+ <-r.Context().Done()
+ }))
+ defer server.Close()
+ input := &http.Client{Timeout: time.Minute}
+ api, err := New(Config{URI: server.URL, HTTPClient: input, RequestTimeout: 50 * time.Millisecond, RecoveryTimeout: time.Second})
+ require.NoError(t, err)
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ defer cancel()
+ err = api.CreateDatabase(ctx, "analytics", nil)
+ require.Error(t, err)
+ require.True(t, IsMutationOutcomeUncertain(err))
+ require.NoError(t, ctx.Err(), "the configured request timeout must expire first")
+ require.Equal(t, int32(1), calls.Load(), "an uncertain mutation must never be blindly retried")
+ require.Equal(t, time.Minute, input.Timeout, "do not alter the caller's shared HTTP client")
+ require.Equal(t, time.Second, api.RecoveryTimeout())
+}
+
+func TestAPIErrorsExposeOnlySafeRequestIDs(t *testing.T) {
+ const requestID = "89f29f74-3a2a-43b9-b698-f24e6e9c872c"
+ for _, test := range []struct {
+ name, header, token, want string
+ }{
+ {name: "correlation", header: requestID, want: requestID},
+ {name: "arbitrary response text", header: "secret-must-not-be-shown"},
+ {name: "echoed bearer", header: requestID, token: requestID},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("X-Acs-Request-Id", test.header)
+ w.WriteHeader(http.StatusForbidden)
+ _, _ = w.Write([]byte(`{"code":403,"StatusCode":503,"message":"secret-response-body"}`))
+ }))
+ defer server.Close()
+ api, err := New(Config{URI: server.URL, Token: test.token})
+ require.NoError(t, err)
+ _, err = api.GetDatabase(context.Background(), "analytics")
+ require.Error(t, err)
+ var apiErr *APIError
+ require.ErrorAs(t, err, &apiErr)
+ require.Equal(t, http.StatusForbidden, apiErr.StatusCode, "the error body must not override the HTTP outcome")
+ require.False(t, IsMutationOutcomeUncertain(err))
+ require.Equal(t, test.want, apiErr.RequestID)
+ require.NotContains(t, err.Error(), "secret")
+ if test.want == "" {
+ require.NotContains(t, err.Error(), requestID)
+ } else {
+ require.Contains(t, err.Error(), requestID)
+ }
+ })
+ }
+}
diff --git a/internal/client/dlf_auth.go b/internal/client/dlf_auth.go
index e9e5cd5..89ed5d4 100644
--- a/internal/client/dlf_auth.go
+++ b/internal/client/dlf_auth.go
@@ -329,9 +329,10 @@ func newDLFCredentialProvider(config DLFConfig, httpClient *http.Client) (dlfCre
return nil, errors.New("DLF ECS metadata URL must be an absolute HTTP(S) URL")
}
loader = &ecsDLFTokenLoader{
- metadataURL: strings.TrimSpace(metadataURL),
+ metadataURL: strings.TrimRight(metadataURL, "/") + "/",
roleName: strings.TrimSpace(config.ECSRoleName),
httpClient: httpClient,
+ now: config.Now,
}
}
@@ -415,7 +416,10 @@ type ecsDLFTokenLoader struct {
roleName string
httpClient *http.Client
- mu sync.Mutex
+ mu sync.Mutex
+ token string
+ tokenExpiresAt time.Time
+ now func() time.Time
}
func (l *ecsDLFTokenLoader) Load(ctx context.Context) (dlfCredentials, error) {
@@ -446,30 +450,81 @@ func (l *ecsDLFTokenLoader) Load(ctx context.Context) (dlfCredentials, error) {
return credentials, nil
}
+const ecsMetadataTokenTTL = 6 * time.Hour
+
+func (l *ecsDLFTokenLoader) metadataToken(ctx context.Context) (string, error) {
+ now := time.Now()
+ if l.now != nil {
+ now = l.now()
+ }
+ if l.token != "" && now.Before(l.tokenExpiresAt.Add(-time.Minute)) {
+ return l.token, nil
+ }
+ endpoint, err := url.Parse(l.metadataURL)
+ if err != nil {
+ return "", errors.New("invalid ECS metadata endpoint")
+ }
+ endpoint.Path = "/latest/api/token"
+ endpoint.RawPath, endpoint.RawQuery, endpoint.Fragment = "", "", ""
+ token, _, err := l.metadataRequest(ctx, http.MethodPut, endpoint.String(), "", 4<<10)
+ if err != nil {
+ return "", fmt.Errorf("obtain ECS metadata token: %w", err)
+ }
+ token = strings.TrimSpace(token)
+ if token == "" || strings.IndexFunc(token, func(r rune) bool { return r < 0x21 || r > 0x7e }) >= 0 {
+ return "", errors.New("ECS metadata returned an invalid access token")
+ }
+ l.token, l.tokenExpiresAt = token, now.Add(ecsMetadataTokenTTL)
+
+ return token, nil
+}
+
func (l *ecsDLFTokenLoader) getText(ctx context.Context, endpoint string, limit int64) (string, error) {
- request, err := http.NewRequestWithContext(ctx, http.MethodGet, endpoint, nil)
+ for attempt := 0; attempt < 2; attempt++ {
+ token, err := l.metadataToken(ctx)
+ if err != nil {
+ return "", err
+ }
+ contents, status, err := l.metadataRequest(ctx, http.MethodGet, endpoint, token, limit)
+ if attempt == 0 && (status == http.StatusUnauthorized || status == http.StatusForbidden) {
+ l.token = ""
+
+ continue
+ }
+
+ return contents, err
+ }
+
+ return "", errors.New("ECS metadata authentication failed")
+}
+
+func (l *ecsDLFTokenLoader) metadataRequest(ctx context.Context, method, endpoint, token string, limit int64) (string, int, error) {
+ request, err := http.NewRequestWithContext(ctx, method, endpoint, nil)
if err != nil {
- return "", errors.New("create metadata request")
+ return "", 0, errors.New("create metadata request")
+ }
+ if method == http.MethodPut {
+ request.Header.Set("X-aliyun-ecs-metadata-token-ttl-seconds", "21600")
+ } else {
+ request.Header.Set("X-aliyun-ecs-metadata-token", token)
}
response, err := l.httpClient.Do(request)
if err != nil {
- return "", errors.New("call metadata service")
+ return "", 0, errors.New("call metadata service")
}
defer response.Body.Close()
if response.StatusCode < http.StatusOK || response.StatusCode >= http.StatusMultipleChoices {
- _, _ = io.Copy(io.Discard, io.LimitReader(response.Body, limit))
-
- return "", fmt.Errorf("metadata service returned HTTP %d", response.StatusCode)
+ return "", response.StatusCode, fmt.Errorf("metadata service returned HTTP %d", response.StatusCode)
}
contents, err := io.ReadAll(io.LimitReader(response.Body, limit+1))
if err != nil {
- return "", errors.New("read metadata response")
+ return "", response.StatusCode, errors.New("read metadata response")
}
if int64(len(contents)) > limit {
- return "", errors.New("metadata response exceeded size limit")
+ return "", response.StatusCode, errors.New("metadata response exceeded size limit")
}
- return string(contents), nil
+ return string(contents), response.StatusCode, nil
}
func signDLFDefault(method, resourcePath string, query url.Values, body []byte, credentials dlfCredentials, region string, now time.Time) (map[string]string, error) {
diff --git a/internal/client/dlf_auth_test.go b/internal/client/dlf_auth_test.go
index f64431f..8544efc 100644
--- a/internal/client/dlf_auth_test.go
+++ b/internal/client/dlf_auth_test.go
@@ -21,6 +21,7 @@ import (
"context"
"encoding/json"
"errors"
+ "fmt"
"io"
"net/http"
"net/http/httptest"
@@ -271,6 +272,9 @@ func TestECSDLFTokenLoaderDiscoversAndCachesRole(t *testing.T) {
var credentialRequests atomic.Int32
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
switch r.URL.Path {
+ case "/latest/api/token":
+ assert.Equal(t, http.MethodPut, r.Method)
+ _, _ = w.Write([]byte("metadata-token"))
case "/metadata/":
roleRequests.Add(1)
_, _ = w.Write([]byte("role-a\n"))
@@ -495,3 +499,67 @@ func TestUniqueDLFNonceConcurrent(t *testing.T) {
}
assert.Len(t, unique, count)
}
+
+func TestECSMetadataTokenRefreshAndRequiredMode(t *testing.T) {
+ now := time.Date(2026, 9, 5, 0, 0, 0, 0, time.UTC)
+ var tokenRequests, roleRequests atomic.Int32
+ var rejectToken atomic.Bool
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/latest/api/token" {
+ require.Equal(t, http.MethodPut, r.Method)
+ require.Equal(t, "21600", r.Header.Get("X-aliyun-ecs-metadata-token-ttl-seconds"))
+ count := tokenRequests.Add(1)
+ _, _ = fmt.Fprintf(w, "token-%d", count)
+
+ return
+ }
+ expected := fmt.Sprintf("token-%d", tokenRequests.Load())
+ if r.Header.Get("X-aliyun-ecs-metadata-token") != expected || rejectToken.Swap(false) {
+ w.WriteHeader(http.StatusUnauthorized)
+ _, _ = w.Write([]byte("metadata-secret-must-not-escape"))
+
+ return
+ }
+ if r.URL.Path == "/metadata/" {
+ roleRequests.Add(1)
+ _, _ = w.Write([]byte("role-a"))
+
+ return
+ }
+ writeJSON(t, w, dlfCredentials{AccessKeyID: "id", AccessKeySecret: "secret", SecurityToken: "sts"})
+ }))
+ defer server.Close()
+ loader := &ecsDLFTokenLoader{metadataURL: server.URL + "/metadata/", httpClient: noRedirectHTTPClient(server.Client()), now: func() time.Time { return now }}
+ for range 2 {
+ creds, err := loader.Load(context.Background())
+ require.NoError(t, err)
+ require.Equal(t, "sts", creds.SecurityToken)
+ }
+ require.Equal(t, int32(1), tokenRequests.Load())
+ now = now.Add(ecsMetadataTokenTTL)
+ _, err := loader.Load(context.Background())
+ require.NoError(t, err)
+ require.Equal(t, int32(2), tokenRequests.Load())
+ rejectToken.Store(true)
+ _, err = loader.Load(context.Background())
+ require.NoError(t, err)
+ require.Equal(t, int32(3), tokenRequests.Load())
+ require.Equal(t, int32(1), roleRequests.Load())
+}
+
+func TestECSMetadataTokenFailureDoesNotDowngrade(t *testing.T) {
+ var ordinaryRequests atomic.Int32
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodPut {
+ ordinaryRequests.Add(1)
+ }
+ w.WriteHeader(http.StatusForbidden)
+ _, _ = w.Write([]byte("metadata-secret-must-not-escape"))
+ }))
+ defer server.Close()
+ loader := &ecsDLFTokenLoader{metadataURL: server.URL + "/metadata/", httpClient: server.Client()}
+ _, err := loader.Load(context.Background())
+ require.Error(t, err)
+ require.NotContains(t, err.Error(), "metadata-secret-must-not-escape")
+ require.Zero(t, ordinaryRequests.Load())
+}
diff --git a/internal/provider/acceptance_test.go b/internal/provider/acceptance_test.go
index 0fe720d..430bba1 100644
--- a/internal/provider/acceptance_test.go
+++ b/internal/provider/acceptance_test.go
@@ -209,7 +209,7 @@ resource "paimon_table" "events" {
name = "events"
fields = [%s
]
- primary_keys = ["id"]
+ options = { "primary-key" = "id" }
comment = %q
}
`, uri, owner, fields, comment)
@@ -250,7 +250,8 @@ resource "paimon_table" "events" {
nullable = true
}%s
]
- primary_keys = ["id"]
+ options = { "primary-key" = "id" }
+ allow_replacement = true
}
`, uri, keyType, requiredField)
}
@@ -261,6 +262,8 @@ type acceptanceCatalog struct {
table *client.Table
databaseCreates int
tableCreates int
+ tableDeletes int
+ tableAlters int
nextTableFieldID int
}
@@ -310,6 +313,19 @@ func (c *acceptanceCatalog) ServeHTTP(w http.ResponseWriter, r *http.Request) {
Schema client.Schema `json:"schema"`
}
_ = json.NewDecoder(r.Body).Decode(&request)
+ if primaryKey, configured := request.Schema.Options["primary-key"]; configured {
+ if len(request.Schema.PrimaryKeys) != 0 {
+ http.Error(w, "duplicate DDL and option primary keys", http.StatusBadRequest)
+
+ return
+ }
+ for _, key := range strings.Split(primaryKey, ",") {
+ if key = strings.TrimSpace(key); key != "" {
+ request.Schema.PrimaryKeys = append(request.Schema.PrimaryKeys, key)
+ }
+ }
+ delete(request.Schema.Options, "primary-key")
+ }
c.tableCreates++
c.table = &client.Table{ID: "table-1", Database: request.Identifier.Database, Name: request.Identifier.Object, SchemaID: 1, Schema: request.Schema}
c.nextTableFieldID = nextAcceptanceFieldID(request.Schema.Fields)
@@ -322,6 +338,7 @@ func (c *acceptanceCatalog) ServeHTTP(w http.ResponseWriter, r *http.Request) {
}
writeAcceptanceJSON(w, c.table)
case r.URL.Path == "/v1/catalog/databases/analytics/tables/events" && r.Method == http.MethodPost:
+ c.tableAlters++
var request struct {
Changes []map[string]json.RawMessage `json:"changes"`
}
@@ -330,6 +347,18 @@ func (c *acceptanceCatalog) ServeHTTP(w http.ResponseWriter, r *http.Request) {
var action string
_ = json.Unmarshal(change["action"], &action)
switch action {
+ case "setOption":
+ var key, value string
+ _ = json.Unmarshal(change["key"], &key)
+ _ = json.Unmarshal(change["value"], &value)
+ if c.table.Schema.Options == nil {
+ c.table.Schema.Options = make(map[string]string)
+ }
+ c.table.Schema.Options[key] = value
+ case "removeOption":
+ var key string
+ _ = json.Unmarshal(change["key"], &key)
+ delete(c.table.Schema.Options, key)
case "updateComment":
_ = json.Unmarshal(change["comment"], &c.table.Schema.Comment)
case "addColumn":
@@ -362,6 +391,7 @@ func (c *acceptanceCatalog) ServeHTTP(w http.ResponseWriter, r *http.Request) {
c.table.SchemaID++
w.WriteHeader(http.StatusOK)
case r.URL.Path == "/v1/catalog/databases/analytics/tables/events" && r.Method == http.MethodDelete:
+ c.tableDeletes++
c.table = nil
w.WriteHeader(http.StatusNoContent)
default:
diff --git a/internal/provider/api_contract_test.go b/internal/provider/api_contract_test.go
new file mode 100644
index 0000000..d29a672
--- /dev/null
+++ b/internal/provider/api_contract_test.go
@@ -0,0 +1,504 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information regarding
+// copyright ownership. The ASF licenses this file to You under the Apache
+// License, Version 2.0 (the "License"); you may not use this file except in
+// compliance with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package provider
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "regexp"
+ "strconv"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/terraform-provider-paimon/internal/client"
+ "github.com/hashicorp/terraform-plugin-framework/diag"
+ frameworkprovider "github.com/hashicorp/terraform-plugin-framework/provider"
+ "github.com/hashicorp/terraform-plugin-framework/resource"
+ "github.com/hashicorp/terraform-plugin-framework/tfsdk"
+ "github.com/hashicorp/terraform-plugin-framework/types"
+ accresource "github.com/hashicorp/terraform-plugin-testing/helper/resource"
+ "github.com/hashicorp/terraform-plugin-testing/plancheck"
+ "github.com/stretchr/testify/require"
+)
+
+func TestProviderRejectsMixedAuthentication(t *testing.T) {
+ ctx := context.Background()
+ p := &paimonProvider{}
+ var schema frameworkprovider.SchemaResponse
+ p.Schema(ctx, frameworkprovider.SchemaRequest{}, &schema)
+ model := paimonProviderModel{Headers: types.MapNull(types.StringType), URI: types.StringValue("http://localhost:8080"), TokenProvider: types.StringValue("bear"), Token: types.StringValue("dummy-token"), DLFTokenLoader: types.StringValue("ecs")}
+ state := tfsdk.State{Schema: schema.Schema}
+ require.False(t, state.Set(ctx, &model).HasError())
+ var response frameworkprovider.ConfigureResponse
+ p.Configure(ctx, frameworkprovider.ConfigureRequest{Config: tfsdk.Config{Schema: schema.Schema, Raw: state.Raw}}, &response)
+ require.True(t, response.Diagnostics.HasError(), "explicit bearer plus ECS DLF credentials must be rejected")
+}
+
+func TestPolicyCreationAcceptsJavaCanonicalFieldReference(t *testing.T) {
+ planned := `{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":99,"name":"tenant","type":"STRING"}},"function":"EQUAL","literals":["APAC"]}`
+ canonical := `{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":1,"name":"tenant","type":"STRING"}},"function":"EQUAL","literals":["APAC"]}`
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeProviderJSON(t, w, client.ConfigResponse{Defaults: map[string]string{"prefix": "catalog"}})
+
+ return
+ }
+ if r.Method == http.MethodGet && r.URL.Path == "/v1/catalog/databases/db/tables/tbl" {
+ writeProviderJSON(t, w, client.Table{Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT"}, {ID: 1, Name: "tenant", Type: "STRING"}}}})
+
+ return
+ }
+ if r.Method == http.MethodPost {
+ w.WriteHeader(http.StatusNoContent)
+
+ return
+ }
+ writeProviderJSON(t, w, client.ListPoliciesResponse{Policies: []client.DataPolicy{{Resource: client.PermissionResource{Type: client.ResourceTypeTable, Database: "db", Table: "tbl"}, Principal: "role:reader", RowFilter: &client.RowFilter{Predicate: canonical}}}})
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL, RecoveryTimeout: 100 * time.Millisecond})
+ require.NoError(t, err)
+ spec := rowFilterSpec(rowFilterResourceModel{Database: types.StringValue("db"), Table: types.StringValue("tbl"), Principal: types.StringValue("role:reader"), Predicate: types.StringValue(planned)})
+ result := createPolicyWithReconciliation(context.Background(), api, spec, false)
+ require.True(t, result.accepted)
+ require.NoError(t, result.err, "Java's name-based field-ref remapping must not be mistaken for a failed mutation")
+}
+
+func TestAccImportPreservesUnconfiguredKeysAndNullability(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ for _, nullable := range []bool{false, true} {
+ t.Run(strconv.FormatBool(nullable), func(t *testing.T) {
+ catalog := &acceptanceCatalog{database: &client.Database{Name: "analytics", Options: map[string]string{}}, table: &client.Table{ID: "table-1", Database: "analytics", Name: "events", SchemaID: 1, Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT NOT NULL"}}, PrimaryKeys: []string{"id"}, Options: map[string]string{"owner": "old"}}}}
+ if nullable {
+ catalog.table.Schema.Fields[0].Type = "BIGINT"
+ catalog.table.Schema.Options["primary-key.nullable"] = "true"
+ }
+
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodPost && r.URL.Path == "/v1/catalog/databases/analytics/tables/events" {
+ body, err := io.ReadAll(r.Body)
+ require.NoError(t, err)
+ require.JSONEq(t, `{"changes":[{"action":"setOption","key":"owner","value":"new"}]}`, string(body))
+ r.Body = io.NopCloser(bytes.NewReader(body))
+ }
+ catalog.ServeHTTP(w, r)
+ }))
+ defer server.Close()
+ config := fmt.Sprintf(`
+provider "paimon" {
+ uri = %q
+ recovery_timeout_seconds = 1
+}
+resource "paimon_table" "events" {
+ database = "analytics"
+ name = "events"
+ fields = [{ name = "id", type = "BIGINT" }]
+ options = { owner = "new" }
+}
+`, server.URL)
+ accresource.Test(t, accresource.TestCase{ProtoV6ProviderFactories: testAccProtoV6ProviderFactories, Steps: []accresource.TestStep{
+ {Config: config, ResourceName: "paimon_table.events", ImportState: true, ImportStateId: "database=analytics&table=events", ImportStatePersist: true},
+ {Config: config},
+ {Config: config, PlanOnly: true},
+ }})
+ })
+ }
+}
+
+func TestAccImportPreservesUnconfiguredKeys(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ catalog := &acceptanceCatalog{database: &client.Database{Name: "analytics", Options: map[string]string{}}, table: &client.Table{ID: "table-1", Database: "analytics", Name: "events", SchemaID: 1, Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT NOT NULL"}}, PrimaryKeys: []string{"id"}, Options: map[string]string{"owner": "old"}}}}
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Method == http.MethodPost && r.URL.Path == "/v1/catalog/databases/analytics/tables/events" {
+ body, err := io.ReadAll(r.Body)
+ require.NoError(t, err)
+ require.JSONEq(t, `{"changes":[{"action":"setOption","key":"owner","value":"new"}]}`, string(body))
+ r.Body = io.NopCloser(bytes.NewReader(body))
+ }
+ catalog.ServeHTTP(w, r)
+ }))
+ defer server.Close()
+ config := fmt.Sprintf(`
+provider "paimon" {
+ uri = %q
+ recovery_timeout_seconds = 1
+}
+resource "paimon_table" "events" {
+ database = "analytics"
+ name = "events"
+ fields = [{ name = "id", type = "BIGINT", nullable = false }]
+ options = { owner = "new" }
+}
+`, server.URL)
+ accresource.Test(t, accresource.TestCase{ProtoV6ProviderFactories: testAccProtoV6ProviderFactories, Steps: []accresource.TestStep{
+ {Config: config, ResourceName: "paimon_table.events", ImportState: true, ImportStateId: "database=analytics&table=events", ImportStatePersist: true},
+ {Config: config},
+ {Config: config, PlanOnly: true},
+ }})
+}
+
+func TestPolicySemanticComparisonPreservesMeaning(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeProviderJSON(t, w, client.ConfigResponse{})
+
+ return
+ }
+ writeProviderJSON(t, w, client.Table{Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT"}, {ID: 1, Name: "tenant", Type: "STRING"}}}})
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL})
+ require.NoError(t, err)
+ leaf := `{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":99,"name":"tenant","type":"INT"}},"function":"EQUAL","literals":["APAC"]}`
+ canonical := strings.ReplaceAll(strings.ReplaceAll(leaf, `"index":99`, `"index":1`), `"type":"INT"`, `"type":"STRING"`)
+ cast := `{"name":"CAST","fieldRef":{"index":99,"name":"tenant","type":"INT"},"type":"STRING"}`
+ for _, test := range []struct {
+ name, kind, left, right string
+ equal, invalid bool
+ }{
+ {name: "name-based reference", kind: client.PolicyTypeRowFilter, left: leaf, right: canonical, equal: true},
+ {name: "single child", kind: client.PolicyTypeRowFilter, left: `{"kind":"COMPOUND","function":"AND","children":[` + leaf + `]}`, right: canonical, equal: true},
+ {name: "changed literal", kind: client.PolicyTypeRowFilter, left: leaf, right: strings.ReplaceAll(canonical, "APAC", "EMEA")},
+ {name: "changed operator", kind: client.PolicyTypeRowFilter, left: leaf, right: strings.ReplaceAll(canonical, "EQUAL", "NOT_EQUAL")},
+ {name: "cast target alias", kind: client.PolicyTypeColumnMasking, left: strings.ReplaceAll(cast, `},"type":"STRING"`, `},"type":"INTEGER"`), right: strings.ReplaceAll(cast, `},"type":"STRING"`, `},"type":"INT"`), equal: true},
+ {name: "cast target nullability", kind: client.PolicyTypeColumnMasking, left: cast, right: strings.ReplaceAll(cast, `},"type":"STRING"`, `},"type":"STRING NOT NULL"`)},
+ {name: "cast target", kind: client.PolicyTypeColumnMasking, left: cast, right: strings.ReplaceAll(cast, `},"type":"STRING"`, `},"type":"BIGINT"`)},
+ {name: "string transform inputs", kind: client.PolicyTypeColumnMasking, left: `{"name":"CONCAT","inputs":[{"index":99,"name":"tenant","type":"INT"},"!"]}`, right: `{"name":"CONCAT","inputs":[{"index":1,"name":"tenant","type":"STRING"},"!"]}`, equal: true},
+ {name: "changed reference name", kind: client.PolicyTypeRowFilter, left: leaf, right: strings.ReplaceAll(canonical, `"name":"tenant"`, `"name":"id"`)},
+ {name: "unknown field", kind: client.PolicyTypeRowFilter, left: leaf, right: strings.ReplaceAll(canonical, `"name":"tenant"`, `"name":"absent"`), invalid: true},
+ {name: "unknown compound attribute", kind: client.PolicyTypeRowFilter, left: `{"kind":"COMPOUND","function":"AND","children":[` + leaf + `],"extra":true}`, right: canonical},
+ {name: "literal metadata is meaningful", kind: client.PolicyTypeRowFilter, left: strings.ReplaceAll(leaf, `["APAC"]`, `[{"name":"tenant","type":"INT","index":99}]`), right: strings.ReplaceAll(canonical, `["APAC"]`, `[{"name":"tenant","type":"STRING","index":1}]`)},
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ equal, err := equivalentPolicyContent(context.Background(), api, policySpec{database: "db", table: "tbl", policyType: test.kind}, test.left, test.right)
+ require.Equal(t, test.invalid, err != nil)
+ require.Equal(t, test.equal, equal)
+ })
+ }
+}
+
+func TestAccPrimaryKeyOptionLifecycle(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ catalog := &acceptanceCatalog{database: &client.Database{Name: "analytics", Options: map[string]string{}}}
+ server := httptest.NewServer(catalog)
+ defer server.Close()
+ config := func(option string, allowed bool) string {
+ return fmt.Sprintf(`
+provider "paimon" { uri = %q }
+resource "paimon_table" "events" {
+ database = "analytics"
+ name = "events"
+ fields = [{name="id",type="BIGINT"},{name="tenant",type="STRING"}]
+ allow_replacement = %t
+ options = { %s }
+}
+`, server.URL, allowed, option)
+ }
+ initial := config(`"primary-key"=" , id , ,tenant, "`, false)
+ removed := config("", true)
+ accresource.Test(t, accresource.TestCase{ProtoV6ProviderFactories: testAccProtoV6ProviderFactories, Steps: []accresource.TestStep{
+ {Config: initial, Check: accresource.ComposeTestCheckFunc(accresource.TestCheckResourceAttr("paimon_table.events", "primary_keys.0", "id"), accresource.TestCheckResourceAttr("paimon_table.events", "primary_keys.1", "tenant"), accresource.TestCheckResourceAttr("paimon_table.events", "fields.0.nullable", "false"), accresource.TestCheckResourceAttr("paimon_table.events", "options.primary-key", " , id , ,tenant, "))},
+ {Config: initial, PlanOnly: true},
+ {Config: config("", false), PlanOnly: true, ExpectError: regexp.MustCompile("Destructive table change is disabled")},
+ {Config: removed, ConfigPlanChecks: accresource.ConfigPlanChecks{PreApply: []plancheck.PlanCheck{plancheck.ExpectResourceAction("paimon_table.events", plancheck.ResourceActionDestroyBeforeCreate)}}, Check: accresource.TestCheckResourceAttr("paimon_table.events", "primary_keys.#", "0")},
+ {Config: removed, PlanOnly: true},
+ }})
+ require.Equal(t, 2, catalog.tableCreates)
+ require.Zero(t, catalog.tableAlters, "primary-key must never be changed through ALTER options")
+}
+
+func TestAccPolicyJavaCanonicalization(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ canonicalPredicate := `{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":1,"name":"tenant","type":"STRING"}},"function":"EQUAL","literals":["APAC"]}`
+ canonicalTransform := `{"name":"UPPER","inputs":[{"index":1,"name":"tenant","type":"STRING"}]}`
+ catalog := &managementAcceptanceCatalog{policies: make(map[string]client.DataPolicy), canonicalize: func(p client.PolicyRequest) client.PolicyRequest {
+ if p.RowFilter != nil {
+ p.RowFilter.Predicate = canonicalPredicate
+ }
+ if p.ColumnMask != nil {
+ p.ColumnMask.Transform = canonicalTransform
+ }
+
+ return p
+ }}
+ server := httptest.NewServer(catalog)
+ defer server.Close()
+ config := func(predicate, transform string) string {
+ return fmt.Sprintf(`
+provider "paimon" { uri = %q }
+%s
+resource "paimon_row_filter" "region" {
+ database=paimon_table.events.database
+ table=paimon_table.events.name
+ principal="role:analyst"
+ predicate=%q
+}
+resource "paimon_column_mask" "secret" {
+ database=paimon_table.events.database
+ table=paimon_table.events.name
+ principal="role:analyst"
+ column="secret"
+ transform=%q
+}
+`, server.URL, managementTableConfig("analytics"), predicate, transform)
+ }
+ oldReference := strings.ReplaceAll(strings.ReplaceAll(canonicalPredicate, `"index":1`, `"index":99`), `"type":"STRING"`, `"type":"INT"`)
+ initial := config(`{"kind":"COMPOUND","function":"AND","children":[`+oldReference+`]}`, strings.ReplaceAll(canonicalTransform, `"index":1`, `"index":99`))
+ canonical := config(canonicalPredicate, canonicalTransform)
+ accresource.Test(t, accresource.TestCase{ProtoV6ProviderFactories: testAccProtoV6ProviderFactories, Steps: []accresource.TestStep{
+ {Config: initial},
+ {Config: initial, PlanOnly: true},
+ {Config: canonical},
+ {Config: canonical, PlanOnly: true},
+ }})
+ require.Equal(t, 4, catalog.policyMutations, "two creates plus final deletes; equivalent updates must never detach policies")
+}
+
+func TestAccTableWithDeferredSchema(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ catalog := &acceptanceCatalog{database: &client.Database{Name: "analytics", Options: map[string]string{}}}
+ server := httptest.NewServer(catalog)
+ defer server.Close()
+ config := fmt.Sprintf(`
+provider "paimon" { uri = %q }
+resource "terraform_data" "schema" { input = [{name="id",type="BIGINT"}] }
+resource "terraform_data" "keys" { input = "id" }
+resource "paimon_table" "events" {
+ database="analytics"
+ name="events"
+ fields=terraform_data.schema.output
+ options={ "primary-key"=terraform_data.keys.output }
+}
+`, server.URL)
+ accresource.Test(t, accresource.TestCase{ProtoV6ProviderFactories: testAccProtoV6ProviderFactories, Steps: []accresource.TestStep{
+ {Config: config, Check: accresource.ComposeTestCheckFunc(accresource.TestCheckResourceAttr("paimon_table.events", "primary_keys.0", "id"), accresource.TestCheckResourceAttr("paimon_table.events", "fields.0.nullable", "false"))},
+ {Config: config, PlanOnly: true},
+ }})
+}
+
+func TestAccTableWithDeferredPartitionKeyElements(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ catalog := &acceptanceCatalog{database: &client.Database{Name: "analytics", Options: map[string]string{}}}
+ server := httptest.NewServer(catalog)
+ defer server.Close()
+ config := func(revision int, prefix string, allowed bool) string {
+ return fmt.Sprintf(`
+provider "paimon" { uri = %q }
+resource "terraform_data" "partition" {
+ input = { name = "tenant", revision = %d }
+}
+resource "paimon_table" "events" {
+ database = "analytics"
+ name = "events"
+ fields = [{name="tenant",type="STRING"},{name="region",type="STRING"}]
+ partition_keys = [%s terraform_data.partition.output.name]
+ allow_replacement = %t
+}
+`, server.URL, revision, prefix, allowed)
+ }
+ initial := config(1, "", false)
+ nullKey := strings.Replace(config(2, "", true), "terraform_data.partition.output.name", "null", 1)
+ // Changing the dependency makes its output unknown again. Adding a second
+ // partition key requires replacement, with both known and unknown elements.
+ updated := config(2, `"region",`, true)
+ accresource.Test(t, accresource.TestCase{
+ ProtoV6ProviderFactories: testAccProtoV6ProviderFactories,
+ Steps: []accresource.TestStep{
+ {Config: nullKey, PlanOnly: true, ExpectError: regexp.MustCompile("Null List Value")},
+ {
+ Config: initial,
+ Check: accresource.ComposeTestCheckFunc(
+ accresource.TestCheckResourceAttr("paimon_table.events", "partition_keys.#", "1"),
+ accresource.TestCheckResourceAttr("paimon_table.events", "partition_keys.0", "tenant"),
+ accresource.TestCheckResourceAttr("paimon_table.events", "fields.0.nullable", "true"),
+ ),
+ },
+ {Config: initial, PlanOnly: true},
+ {Config: nullKey, PlanOnly: true, ExpectError: regexp.MustCompile("Null List Value")},
+ {
+ Config: config(2, `"region",`, false),
+ PlanOnly: true,
+ ExpectError: regexp.MustCompile("Destructive table change is disabled"),
+ },
+ {
+ Config: updated,
+ ConfigPlanChecks: accresource.ConfigPlanChecks{PreApply: []plancheck.PlanCheck{
+ plancheck.ExpectResourceAction("paimon_table.events", plancheck.ResourceActionDestroyBeforeCreate),
+ }},
+ Check: accresource.ComposeTestCheckFunc(
+ accresource.TestCheckResourceAttr("paimon_table.events", "partition_keys.#", "2"),
+ accresource.TestCheckResourceAttr("paimon_table.events", "partition_keys.0", "region"),
+ accresource.TestCheckResourceAttr("paimon_table.events", "partition_keys.1", "tenant"),
+ ),
+ },
+ {Config: updated, PlanOnly: true},
+ },
+ })
+ require.Equal(t, 2, catalog.tableCreates)
+ require.Zero(t, catalog.tableAlters, "partition changes must replace the table, never alter its schema in place")
+}
+
+func TestAccUnknownTableIdentityRequiresReplacementOptIn(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ for _, test := range []struct {
+ attribute, before, after string
+ }{
+ {"partition_keys", `["tenant"]`, `["region"]`},
+ {"name", `"events"`, `"renamed"`},
+ {"database", `"analytics"`, `"other"`},
+ } {
+ t.Run(test.attribute, func(t *testing.T) {
+ catalog := &acceptanceCatalog{database: &client.Database{Name: "analytics", Options: map[string]string{}}}
+ server := httptest.NewServer(catalog)
+ defer server.Close()
+ config := func(input string, allowed bool) string {
+ database, name, keys := `"analytics"`, `"events"`, `["tenant"]`
+ switch test.attribute {
+ case "database":
+ database = "terraform_data.identity.output"
+ case "name":
+ name = "terraform_data.identity.output"
+ case "partition_keys":
+ keys = "terraform_data.identity.output"
+ }
+
+ return fmt.Sprintf(`
+provider "paimon" { uri = %q }
+resource "terraform_data" "identity" { input = %s }
+resource "paimon_table" "events" {
+ database = %s
+ name = %s
+ fields = [{name="tenant",type="STRING"},{name="region",type="STRING"}]
+ partition_keys = %s
+ allow_replacement = %t
+}
+`, server.URL, input, database, name, keys, allowed)
+ }
+ initial := config(test.before, false)
+ accresource.Test(t, accresource.TestCase{
+ ProtoV6ProviderFactories: testAccProtoV6ProviderFactories,
+ Steps: []accresource.TestStep{
+ {
+ Config: initial,
+ Check: accresource.ComposeTestCheckFunc(
+ accresource.TestCheckResourceAttr("paimon_table.events", "id", "database=analytics&table=events"),
+ accresource.TestCheckResourceAttr("paimon_table.events", "partition_keys.0", "tenant"),
+ ),
+ },
+ {Config: initial, PlanOnly: true},
+ {
+ Config: config(test.after, false),
+ ExpectError: regexp.MustCompile("Destructive table change is disabled"),
+ },
+ {Config: initial, PlanOnly: true},
+ {
+ Config: config(test.after, true),
+ PlanOnly: true,
+ ExpectNonEmptyPlan: true,
+ ConfigPlanChecks: accresource.ConfigPlanChecks{PostApplyPreRefresh: []plancheck.PlanCheck{
+ plancheck.ExpectResourceAction("paimon_table.events", plancheck.ResourceActionDestroyBeforeCreate),
+ }},
+ },
+ },
+ })
+ require.Equal(t, 1, catalog.tableCreates, "an unknown identity must not bypass replacement opt-in")
+ require.Equal(t, 1, catalog.tableDeletes, "only the test's final destroy may delete the table")
+ require.Zero(t, catalog.tableAlters, "rejected or plan-only changes must not alter the table")
+ })
+ }
+}
+
+func TestFieldDefaultConstantsAndRemoval(t *testing.T) {
+ ctx := context.Background()
+ for _, test := range []struct{ kind, value string }{{"BIGINT", "42"}, {"BOOLEAN", "true"}, {"DATE", "2026-01-02"}, {"STRING", ""}, {"STRING", "NULL"}, {"STRING", "1 + 2"}, {"STRING", "CURRENT_TIMESTAMP"}} {
+ t.Run(test.kind+"/"+test.value, func(t *testing.T) {
+ var diags diag.Diagnostics
+ field := tableFieldForTest("value", types.Int64Value(0))
+ field.Type = types.StringValue(test.kind)
+ field.DefaultValue = types.StringValue(test.value)
+ model := tableResourceModel{Fields: fieldsValueFromModels(ctx, []tableFieldModel{field}, &diags), Options: types.MapNull(types.StringType), PartitionKeys: types.ListNull(types.StringType), PrimaryKeys: types.ListNull(types.StringType)}
+ schema := schemaFromResourceModel(ctx, &model, &diags)
+ require.False(t, diags.HasError(), diags)
+ encoded, err := json.Marshal(schema.Fields[0])
+ require.NoError(t, err)
+ var wire map[string]any
+ require.NoError(t, json.Unmarshal(encoded, &wire))
+ require.Equal(t, test.value, wire["defaultValue"], "constants must be transmitted unchanged, including empty strings")
+ after := schema.Fields[0]
+ after.DefaultValue = nil
+ changes, err := tableFieldSchemaChanges(schema.Fields, []client.Field{after}, false, nil)
+ require.NoError(t, err)
+ encoded, err = json.Marshal(changes)
+ require.NoError(t, err)
+ require.JSONEq(t, `[{"action":"updateColumnDefaultValue","fieldNames":["value"],"newDefaultValue":null}]`, string(encoded))
+ })
+ }
+}
+
+func TestPolicyReplacementDoesNotReadUncreatedDestination(t *testing.T) {
+ ctx := context.Background()
+ for _, mask := range []bool{false, true} {
+ t.Run(strconv.FormatBool(mask), func(t *testing.T) {
+ var res resource.Resource
+ var before, after any
+ if mask {
+ res = &columnMaskResource{}
+ old := columnMaskResourceModel{Database: types.StringValue("db"), Table: types.StringValue("old"), Column: types.StringValue("secret"), Principal: types.StringValue("role:reader"), Transform: types.StringValue(`{"name":"FIELD_REF","fieldRef":{"index":0,"name":"tenant","type":"STRING"}}`)}
+ next := old
+ next.Table = types.StringValue("not-yet-created")
+ next.Transform = types.StringValue(`{"name":"FIELD_REF","fieldRef":{"index":0,"name":"other","type":"STRING"}}`)
+ before, after = &old, &next
+ } else {
+ res = &rowFilterResource{}
+ old := rowFilterResourceModel{Database: types.StringValue("db"), Table: types.StringValue("old"), Principal: types.StringValue("role:reader"), Predicate: types.StringValue(`{"kind":"LEAF","transform":{"name":"FIELD_REF","fieldRef":{"index":0,"name":"tenant","type":"STRING"}},"function":"EQUAL","literals":["APAC"]}`)}
+ next := old
+ next.Table = types.StringValue("not-yet-created")
+ next.Predicate = types.StringValue(strings.ReplaceAll(old.Predicate.ValueString(), "tenant", "other"))
+ before, after = &old, &next
+ }
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ state := tfsdk.State{Schema: sr.Schema}
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, before).HasError())
+ require.False(t, plan.Set(ctx, after).HasError())
+ response := resource.ModifyPlanResponse{Plan: plan}
+ res.(resource.ResourceWithModifyPlan).ModifyPlan(ctx, resource.ModifyPlanRequest{State: state, Plan: plan}, &response)
+ require.False(t, response.Diagnostics.HasError(), response.Diagnostics)
+ })
+ }
+}
diff --git a/internal/provider/data_source_database.go b/internal/provider/data_source_database.go
index d86290a..1b2b2d0 100644
--- a/internal/provider/data_source_database.go
+++ b/internal/provider/data_source_database.go
@@ -34,7 +34,7 @@ type databaseDataSource struct {
type databaseDataSourceModel struct {
ID types.String `tfsdk:"id"`
- CatalogID types.String `tfsdk:"catalog_id"`
+ ServerID types.String `tfsdk:"server_id"`
Name types.String `tfsdk:"name"`
Options types.Map `tfsdk:"options"`
Location types.String `tfsdk:"location"`
@@ -57,9 +57,9 @@ func (d *databaseDataSource) Schema(_ context.Context, _ datasource.SchemaReques
resp.Schema = schema.Schema{
Description: "Reads an existing database from a Paimon REST Catalog.",
Attributes: map[string]schema.Attribute{
- "id": schema.StringAttribute{Description: "Terraform identifier, equal to the database name.", Computed: true},
- "catalog_id": schema.StringAttribute{Description: "Server-assigned database identifier.", Computed: true},
- "name": schema.StringAttribute{Description: "Database name.", Required: true},
+ "id": schema.StringAttribute{Description: "Terraform identifier, equal to the database name.", Computed: true},
+ "server_id": schema.StringAttribute{Description: "Server-assigned database identifier.", Computed: true},
+ "name": schema.StringAttribute{Description: "Database name.", Required: true},
"options": schema.MapAttribute{
Description: "All database options returned by the REST Catalog.",
Computed: true,
@@ -92,7 +92,7 @@ func (d *databaseDataSource) Read(ctx context.Context, req datasource.ReadReques
return
}
data.ID = types.StringValue(database.Name)
- data.CatalogID = types.StringValue(database.ID)
+ data.ServerID = types.StringValue(database.ID)
data.Name = types.StringValue(database.Name)
data.Options = stringMapValue(ctx, database.Options, &resp.Diagnostics)
data.Location = types.StringValue(database.Location)
diff --git a/internal/provider/data_source_table.go b/internal/provider/data_source_table.go
index 4d9df76..c584758 100644
--- a/internal/provider/data_source_table.go
+++ b/internal/provider/data_source_table.go
@@ -34,7 +34,7 @@ type tableDataSource struct {
type tableDataSourceModel struct {
ID types.String `tfsdk:"id"`
- CatalogID types.String `tfsdk:"catalog_id"`
+ ServerID types.String `tfsdk:"server_id"`
Database types.String `tfsdk:"database"`
Name types.String `tfsdk:"name"`
Fields types.List `tfsdk:"fields"`
@@ -92,7 +92,7 @@ func (d *tableDataSource) Read(ctx context.Context, req datasource.ReadRequest,
name = data.Name.ValueString()
}
data.ID = types.StringValue(tableID(database, name))
- data.CatalogID = types.StringValue(table.ID)
+ data.ServerID = types.StringValue(table.ID)
data.Database = types.StringValue(database)
data.Name = types.StringValue(name)
data.Fields = fieldsValueFromRemote(ctx, table.Schema.Fields, &resp.Diagnostics)
diff --git a/internal/provider/documentation_test.go b/internal/provider/documentation_test.go
new file mode 100644
index 0000000..8fae6a9
--- /dev/null
+++ b/internal/provider/documentation_test.go
@@ -0,0 +1,151 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package provider
+
+import (
+ "context"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/hashicorp/terraform-plugin-framework/providerserver"
+ "github.com/hashicorp/terraform-plugin-go/tfprotov6"
+ "github.com/hashicorp/terraform-plugin-go/tftypes"
+ "github.com/stretchr/testify/require"
+)
+
+// Compare hand-written field references with the schema actually served over
+// the Terraform protocol. Prose examples alone must not count as API coverage.
+func TestDocumentationSchema(t *testing.T) {
+ server := providerserver.NewProtocol6(New("docs")())()
+ response, err := server.GetProviderSchema(context.Background(), &tfprotov6.GetProviderSchemaRequest{})
+ require.NoError(t, err)
+ require.Empty(t, response.Diagnostics)
+ pages := map[string]*tfprotov6.Schema{"index.md": response.Provider}
+ for name, schema := range response.ResourceSchemas {
+ pages[filepath.Join("resources", strings.TrimPrefix(name, "paimon_")+".md")] = schema
+ }
+ for name, schema := range response.DataSourceSchemas {
+ pages[filepath.Join("data-sources", strings.TrimPrefix(name, "paimon_")+".md")] = schema
+ }
+ for page, schema := range pages {
+ t.Run(page, func(t *testing.T) {
+ data, err := os.ReadFile(filepath.Join("..", "..", "docs", page))
+ require.NoError(t, err, "every registered API must have a reference page")
+ text := string(data)
+ require.Equal(t, 1, strings.Count(text, "<!-- schema:begin -->"))
+ require.Equal(t, 1, strings.Count(text, "<!-- schema:end -->"))
+ _, section, _ := strings.Cut(text, "<!-- schema:begin -->")
+ section, _, found := strings.Cut(section, "<!-- schema:end -->")
+ require.True(t, found)
+ require.Empty(t, schema.Block.BlockTypes, "document new nested blocks explicitly before extending this check")
+ expected := make(map[string]documentedAttribute)
+ collectDocumentedAttributes(t, schema.Block.Attributes, "", expected)
+ // Exercise both checkout formats on every OS, including Unix runners.
+ section = strings.ReplaceAll(section, "\r\n", "\n")
+ for name, document := range map[string]string{
+ "LF": section,
+ "CRLF": strings.ReplaceAll(section, "\n", "\r\n"),
+ } {
+ t.Run(name, func(t *testing.T) {
+ actual := parseDocumentedAttributes(t, document)
+ require.Equal(t, expected, actual, "update docs/%s to match the public schema", page)
+ })
+ }
+ })
+ }
+}
+
+type documentedAttribute struct{ Type, Mode, Sensitive string }
+
+func parseDocumentedAttributes(t *testing.T, section string) map[string]documentedAttribute {
+ t.Helper()
+ attributes := make(map[string]documentedAttribute)
+ for _, line := range strings.Split(section, "\n") {
+ line = strings.TrimSpace(line)
+ if !strings.HasPrefix(line, "| `") {
+ continue
+ }
+ columns := strings.Split(strings.Trim(line, "|"), "|")
+ require.Len(t, columns, 5, "schema rows have attribute, type, mode, sensitive and meaning columns")
+ for i := range columns {
+ columns[i] = strings.Trim(strings.TrimSpace(columns[i]), "`")
+ }
+ require.NotEmpty(t, columns[4], "every attribute needs a meaning/default description")
+ _, duplicate := attributes[columns[0]]
+ require.False(t, duplicate, "duplicate documented attribute: %s", columns[0])
+ attributes[columns[0]] = documentedAttribute{Type: columns[1], Mode: columns[2], Sensitive: columns[3]}
+ }
+
+ return attributes
+}
+
+func collectDocumentedAttributes(t *testing.T, attributes []*tfprotov6.SchemaAttribute, prefix string, result map[string]documentedAttribute) {
+ t.Helper()
+ for _, attribute := range attributes {
+ mode := "Computed"
+ switch {
+ case attribute.Required:
+ mode = "Required"
+ case attribute.Optional && attribute.Computed:
+ mode = "Optional + computed"
+ case attribute.Optional:
+ mode = "Optional"
+ }
+ sensitive := "No"
+ if attribute.Sensitive {
+ sensitive = "Yes"
+ }
+ name := prefix + attribute.Name
+ result[name] = documentedAttribute{Type: documentationType(t, attribute.ValueType()), Mode: mode, Sensitive: sensitive}
+ if nested := attribute.NestedType; nested != nil {
+ suffix := "."
+ if nested.Nesting != tfprotov6.SchemaObjectNestingModeSingle {
+ suffix = "[]."
+ }
+ collectDocumentedAttributes(t, nested.Attributes, name+suffix, result)
+ }
+ }
+}
+
+func documentationType(t *testing.T, value tftypes.Type) string {
+ t.Helper()
+ switch {
+ case value.Is(tftypes.String):
+ return "string"
+ case value.Is(tftypes.Number):
+ return "number"
+ case value.Is(tftypes.Bool):
+ return "bool"
+ }
+ switch typed := value.(type) {
+ case tftypes.List:
+ return "list(" + documentationType(t, typed.ElementType) + ")"
+ case tftypes.Set:
+ return "set(" + documentationType(t, typed.ElementType) + ")"
+ case tftypes.Map:
+ return "map(" + documentationType(t, typed.ElementType) + ")"
+ case tftypes.Object:
+ return "object"
+ default:
+ t.Fatalf("document the new schema type %s", value)
+
+ return ""
+ }
+}
diff --git a/internal/provider/helpers.go b/internal/provider/helpers.go
index cd91fea..405f99b 100644
--- a/internal/provider/helpers.go
+++ b/internal/provider/helpers.go
@@ -31,8 +31,7 @@ import (
)
const (
- mutationRecoveryTimeout = 5 * time.Second
- mutationReadRetryDelay = 50 * time.Millisecond
+ mutationReadRetryDelay = 50 * time.Millisecond
)
func clientFromProviderData(data any, target **client.Client, diags *diag.Diagnostics, kind string) {
@@ -124,10 +123,10 @@ func equivalentJSON(left, right string) bool {
return reflect.DeepEqual(leftValue, rightValue)
}
-func mutationRecoveryContext(ctx context.Context) (context.Context, context.CancelFunc) {
+func mutationRecoveryContext(ctx context.Context, api *client.Client) (context.Context, context.CancelFunc) {
// A canceled apply context can still follow a mutation that committed remotely.
// Keep request-scoped values, but give reconciliation and repair a fresh bound.
- return context.WithTimeout(context.WithoutCancel(ctx), mutationRecoveryTimeout)
+ return context.WithTimeout(context.WithoutCancel(ctx), api.RecoveryTimeout())
}
func retryLookupUntil[T any](ctx context.Context, lookup func(context.Context) (T, bool, error), ready func(T) bool) (T, bool, bool, error) {
@@ -197,3 +196,20 @@ func diffOptions(before, after map[string]string) ([]string, map[string]string)
return removals, updates
}
+
+// optionsConverged checks both halves of an update. A missing key is not an
+// empty value, and a removed key must no longer be present in the catalog.
+func optionsConverged(remote, planned map[string]string, removals []string) bool {
+ for key, value := range planned {
+ if observed, exists := remote[key]; !exists || observed != value {
+ return false
+ }
+ }
+ for _, key := range removals {
+ if _, exists := remote[key]; exists {
+ return false
+ }
+ }
+
+ return true
+}
diff --git a/internal/provider/import_acceptance_test.go b/internal/provider/import_acceptance_test.go
new file mode 100644
index 0000000..3fe15a6
--- /dev/null
+++ b/internal/provider/import_acceptance_test.go
@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information regarding
+// copyright ownership. The ASF licenses this file to You under the Apache
+// License, Version 2.0 (the "License"); you may not use this file except in
+// compliance with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package provider
+
+import (
+ "fmt"
+ "net/http/httptest"
+ "os"
+ "testing"
+
+ "github.com/apache/terraform-provider-paimon/internal/client"
+ accresource "github.com/hashicorp/terraform-plugin-testing/helper/resource"
+ "github.com/hashicorp/terraform-plugin-testing/plancheck"
+ "github.com/stretchr/testify/require"
+)
+
+func TestAccImportMatchingImmutableOptionPreservesTable(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ catalog := &acceptanceCatalog{
+ database: &client.Database{ID: "db-1", Name: "analytics", Options: map[string]string{}},
+ table: &client.Table{ID: "table-1", Database: "analytics", Name: "events", SchemaID: 1, Schema: client.Schema{
+ Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT NOT NULL"}},
+ PrimaryKeys: []string{"id"}, Options: map[string]string{"merge-engine": "deduplicate"},
+ }},
+ }
+ server := httptest.NewServer(catalog)
+ defer server.Close()
+ config := fmt.Sprintf(`
+provider "paimon" { uri = %q }
+resource "paimon_table" "events" {
+ database = "analytics"
+ name = "events"
+ fields = [{name="id",type="BIGINT",nullable=false}]
+ options = {
+ "primary-key" = "id"
+ "merge-engine" = "deduplicate"
+ }
+}
+`, server.URL)
+ accresource.Test(t, accresource.TestCase{
+ ProtoV6ProviderFactories: testAccProtoV6ProviderFactories,
+ Steps: []accresource.TestStep{
+ {Config: config, ResourceName: "paimon_table.events", ImportState: true, ImportStateId: "database=analytics&table=events", ImportStatePersist: true},
+ {Config: config, PlanOnly: true, ExpectNonEmptyPlan: true, ConfigPlanChecks: accresource.ConfigPlanChecks{PostApplyPreRefresh: []plancheck.PlanCheck{plancheck.ExpectResourceAction("paimon_table.events", plancheck.ResourceActionUpdate)}}},
+ {Config: config},
+ {Config: config, PlanOnly: true},
+ },
+ })
+ require.Zero(t, catalog.tableCreates)
+ require.Zero(t, catalog.tableAlters, "taking ownership of the same immutable option must not alter the catalog")
+}
diff --git a/internal/provider/management_acceptance_test.go b/internal/provider/management_acceptance_test.go
new file mode 100644
index 0000000..41ad155
--- /dev/null
+++ b/internal/provider/management_acceptance_test.go
@@ -0,0 +1,200 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information regarding
+// copyright ownership. The ASF licenses this file to You under the Apache
+// License, Version 2.0 (the "License"); you may not use this file except in
+// compliance with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package provider
+
+import (
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "strings"
+ "sync"
+ "testing"
+
+ "github.com/apache/terraform-provider-paimon/internal/client"
+ "github.com/hashicorp/terraform-plugin-testing/helper/resource"
+ "github.com/hashicorp/terraform-plugin-testing/plancheck"
+ "github.com/stretchr/testify/require"
+)
+
+// The fixture exercises Terraform ownership and state mapping. Live server
+// compatibility is covered separately by TestAccRealCatalog.
+type managementAcceptanceCatalog struct {
+ catalog acceptanceCatalog
+ mu sync.Mutex
+ permissions []client.PermissionAssignment
+ policies map[string]client.DataPolicy
+ canonicalize func(client.PolicyRequest) client.PolicyRequest
+ policyMutations int
+}
+
+func (c *managementAcceptanceCatalog) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ if !strings.Contains(r.URL.Path, "/permissions") && !strings.Contains(r.URL.Path, "/policies") {
+ c.catalog.ServeHTTP(w, r)
+
+ return
+ }
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ w.Header().Set("Content-Type", "application/json")
+ if r.Method == http.MethodPost && strings.Contains(r.URL.Path, "/policies") {
+ c.policyMutations++
+ }
+ switch {
+ case strings.HasSuffix(r.URL.Path, "/permissions/grant"):
+ var p client.PermissionAssignment
+ if json.NewDecoder(r.Body).Decode(&p) != nil {
+ w.WriteHeader(http.StatusBadRequest)
+
+ return
+ }
+ c.permissions = []client.PermissionAssignment{p}
+ case strings.HasSuffix(r.URL.Path, "/permissions/revoke"):
+ c.permissions = nil
+ case strings.HasSuffix(r.URL.Path, "/permissions"):
+ writeAcceptanceJSON(w, client.ListPermissionsResponse{Permissions: c.permissions})
+ case strings.HasSuffix(r.URL.Path, "/policies/drop"):
+ var body struct {
+ Type string `json:"type"`
+ Principal string `json:"principal"`
+ Column string `json:"column"`
+ }
+ if json.NewDecoder(r.Body).Decode(&body) != nil {
+ w.WriteHeader(http.StatusBadRequest)
+
+ return
+ }
+ delete(c.policies, body.Type)
+ case r.Method == http.MethodPost:
+ var p client.PolicyRequest
+ if json.NewDecoder(r.Body).Decode(&p) != nil {
+ w.WriteHeader(http.StatusBadRequest)
+
+ return
+ }
+ if c.canonicalize != nil {
+ p = c.canonicalize(p)
+ }
+ kind := client.PolicyTypeRowFilter
+ if p.ColumnMask != nil {
+ kind = client.PolicyTypeColumnMasking
+ }
+ if _, exists := c.policies[kind]; exists {
+ w.WriteHeader(http.StatusConflict)
+
+ return
+ }
+ c.policies[kind] = client.DataPolicy{Resource: client.PermissionResource{Type: client.ResourceTypeTable, Database: "analytics", Table: "events"}, Principal: p.Principal, RowFilter: p.RowFilter, ColumnMask: p.ColumnMask}
+ case r.Method == http.MethodGet:
+ policies := make([]client.DataPolicy, 0, 1)
+ if p, ok := c.policies[r.URL.Query().Get("type")]; ok {
+ policies = append(policies, p)
+ }
+ writeAcceptanceJSON(w, client.ListPoliciesResponse{Policies: policies})
+ default:
+ http.NotFound(w, r)
+ }
+}
+
+func managementResourceConfig(principal string, updated bool) string {
+ region, mask, expiry := "APAC", "***", "null"
+ if updated {
+ region, mask, expiry = "EMEA", "hidden", `"2099-01-01T00:00:00Z"`
+ }
+
+ return fmt.Sprintf(`
+resource "paimon_permission" "read" {
+ resource_type = "TABLE"
+ database = paimon_table.events.database
+ table = paimon_table.events.name
+ access = "SELECT"
+ principal = %q
+ expire_time = %s
+}
+resource "paimon_row_filter" "region" {
+ database = paimon_table.events.database
+ table = paimon_table.events.name
+ principal = paimon_permission.read.principal
+ allow_non_atomic_update = true
+ predicate = jsonencode({
+ kind = "LEAF"
+ transform = { name = "FIELD_REF", fieldRef = { index = 1, name = "tenant", type = "STRING" } }
+ function = "EQUAL"
+ literals = [%q]
+ })
+}
+resource "paimon_column_mask" "secret" {
+ database = paimon_table.events.database
+ table = paimon_table.events.name
+ principal = paimon_permission.read.principal
+ column = "secret"
+ allow_non_atomic_update = true
+ transform = jsonencode({ name = "CONCAT", inputs = [%q] })
+}
+`, principal, expiry, region, mask)
+}
+
+func managementTableConfig(database string) string {
+ return fmt.Sprintf(`
+resource "paimon_database" "analytics" { name = %q }
+resource "paimon_table" "events" {
+ database = paimon_database.analytics.name
+ name = "events"
+ fields = [
+ { name = "id", type = "BIGINT", nullable = false },
+ { name = "tenant", type = "STRING" },
+ { name = "secret", type = "STRING" }
+ ]
+ options = {
+ "primary-key" = "id"
+ "query-auth.enabled" = "true"
+ }
+}
+`, database)
+}
+
+func TestAccManagementResourceLifecycle(t *testing.T) {
+ if os.Getenv("TF_ACC") == "" {
+ t.Skip("set TF_ACC=1 to run Terraform acceptance tests")
+ }
+ catalog := &managementAcceptanceCatalog{policies: make(map[string]client.DataPolicy)}
+ server := httptest.NewServer(catalog)
+ defer server.Close()
+ base := fmt.Sprintf("provider \"paimon\" { uri = %q }\n", server.URL) + managementTableConfig("analytics")
+ initial := base + managementResourceConfig("role:analyst", false)
+ updated := base + managementResourceConfig("role:analyst", true)
+ resource.Test(t, resource.TestCase{
+ ProtoV6ProviderFactories: testAccProtoV6ProviderFactories,
+ Steps: []resource.TestStep{
+ {Config: initial},
+ {Config: initial, PlanOnly: true},
+ {ResourceName: "paimon_permission.read", ImportState: true, ImportStateVerify: true},
+ {ResourceName: "paimon_row_filter.region", ImportState: true, ImportStateVerify: true, ImportStateVerifyIgnore: []string{"allow_non_atomic_update"}},
+ {ResourceName: "paimon_column_mask.secret", ImportState: true, ImportStateVerify: true, ImportStateVerifyIgnore: []string{"allow_non_atomic_update"}},
+ {Config: updated, ConfigPlanChecks: resource.ConfigPlanChecks{PreApply: []plancheck.PlanCheck{
+ plancheck.ExpectResourceAction("paimon_permission.read", plancheck.ResourceActionUpdate),
+ plancheck.ExpectResourceAction("paimon_row_filter.region", plancheck.ResourceActionUpdate),
+ plancheck.ExpectResourceAction("paimon_column_mask.secret", plancheck.ResourceActionUpdate),
+ }}},
+ {Config: updated, PlanOnly: true},
+ },
+ })
+ catalog.mu.Lock()
+ defer catalog.mu.Unlock()
+ require.Empty(t, catalog.policies)
+ require.Empty(t, catalog.permissions)
+}
diff --git a/internal/provider/policy_semantics.go b/internal/provider/policy_semantics.go
new file mode 100644
index 0000000..bb13a27
--- /dev/null
+++ b/internal/provider/policy_semantics.go
@@ -0,0 +1,198 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package provider
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "reflect"
+ "strconv"
+ "strings"
+
+ "github.com/apache/terraform-provider-paimon/internal/client"
+)
+
+// Follow Java's TableQueryAuthResult.remapPredicate and RESTCatalogServerUtils:
+// field references resolve by name; a one-child AND/OR reduces to its child.
+// Only known AST positions are normalized. Literals, cast targets and unknown
+// members retain their meaning and must still compare equal.
+func equivalentPolicyContent(ctx context.Context, api *client.Client, spec policySpec, left, right string) (bool, error) {
+ if equivalentJSON(left, right) {
+ return true, nil
+ }
+ if left == "" || right == "" {
+ return false, nil
+ }
+ var before, after any
+ for _, item := range []struct {
+ text string
+ target *any
+ }{{left, &before}, {right, &after}} {
+ decoder := json.NewDecoder(strings.NewReader(item.text))
+ decoder.UseNumber()
+ if err := decoder.Decode(item.target); err != nil {
+ return false, nil
+ }
+ }
+ var fields []client.Field
+ loaded := false
+ remap := func(ref map[string]any) error {
+ name, ok := ref["name"].(string)
+ if !ok {
+ return errors.New("policy field reference has no name")
+ }
+ if !loaded {
+ if api == nil {
+ return errors.New("a configured catalog is required to resolve policy field references")
+ }
+ table, err := api.GetTable(ctx, spec.database, spec.table)
+ if err != nil {
+ return fmt.Errorf("resolve policy references against table schema: %w", err)
+ }
+ fields, loaded = table.Schema.Fields, true
+ }
+ for index, field := range fields {
+ if field.Name != name {
+ continue
+ }
+ encoded, err := json.Marshal(field.Type)
+ if err != nil {
+ return fmt.Errorf("resolve policy field type: %w", err)
+ }
+ var dataType any
+ decoder := json.NewDecoder(strings.NewReader(string(encoded)))
+ decoder.UseNumber()
+ if err := decoder.Decode(&dataType); err != nil {
+ return fmt.Errorf("resolve policy field type: %w", err)
+ }
+ ref["index"], ref["type"] = json.Number(strconv.Itoa(index)), dataType
+
+ return nil
+ }
+
+ return errors.New("policy field reference does not resolve in the current table schema")
+ }
+ predicate := spec.policyType == client.PolicyTypeRowFilter
+ var err error
+ before, err = normalizePolicyAST(before, predicate, remap)
+ if err != nil {
+ return false, err
+ }
+ after, err = normalizePolicyAST(after, predicate, remap)
+ if err != nil {
+ return false, err
+ }
+
+ return reflect.DeepEqual(before, after), nil
+}
+
+func normalizePolicyAST(value any, predicate bool, remap func(map[string]any) error) (any, error) {
+ node, ok := value.(map[string]any)
+ if !ok {
+ return value, nil
+ }
+ if predicate {
+ switch node["kind"] {
+ case "COMPOUND":
+ children, ok := node["children"].([]any)
+ if !ok {
+ return value, nil
+ }
+ for i, child := range children {
+ normalized, err := normalizePolicyAST(child, true, remap)
+ if err != nil {
+ return nil, err
+ }
+ children[i] = normalized
+ }
+ if len(children) == 1 && len(node) == 3 && (node["function"] == "AND" || node["function"] == "OR") {
+ return children[0], nil
+ }
+ case "LEAF":
+ normalized, err := normalizePolicyAST(node["transform"], false, remap)
+ if err != nil {
+ return nil, err
+ }
+ node["transform"] = normalized
+ }
+
+ return node, nil
+ }
+ switch node["name"] {
+ case "FIELD_REF", "CAST":
+ if ref, ok := node["fieldRef"].(map[string]any); ok {
+ if err := remap(ref); err != nil {
+ return nil, err
+ }
+ }
+ if node["name"] == "CAST" {
+ if target, ok := node["type"].(string); ok && !client.IsCompositeDataType(client.DataType(target)) {
+ // Java serializes atomic aliases such as INTEGER as INT.
+ // Keep the cast target's type and nullability significant.
+ encoded, err := json.Marshal(client.DataType(target))
+ if err != nil {
+ return nil, err
+ }
+ var canonical string
+ if err := json.Unmarshal(encoded, &canonical); err != nil {
+ return nil, err
+ }
+ node["type"] = canonical
+ }
+ }
+ case "CONCAT", "CONCAT_WS", "UPPER", "LOWER", "SUBSTRING", "TRIM":
+ if inputs, ok := node["inputs"].([]any); ok {
+ for _, input := range inputs {
+ if ref, ok := input.(map[string]any); ok {
+ if err := remap(ref); err != nil {
+ return nil, err
+ }
+ }
+ }
+ }
+ }
+
+ return node, nil
+}
+
+func (s policySpec) matchesWithSchema(ctx context.Context, api *client.Client, policy client.DataPolicy) (bool, error) {
+ switch s.policyType {
+ case client.PolicyTypeRowFilter:
+ if policy.RowFilter != nil {
+ return equivalentPolicyContent(ctx, api, s, s.content, policy.RowFilter.Predicate)
+ }
+ case client.PolicyTypeColumnMasking:
+ if policy.ColumnMask != nil && policy.ColumnMask.OnColumn == s.column {
+ return equivalentPolicyContent(ctx, api, s, s.content, policy.ColumnMask.Transform)
+ }
+ }
+
+ return false, nil
+}
+
+func observePolicy(ctx context.Context, api *client.Client, spec policySpec) (policyLookupObservation, error) {
+ policy, found, err := lookupPolicy(ctx, api, spec)
+ if err != nil || !found {
+ return policyLookupObservation{found: found}, err
+ }
+ matches, err := spec.matchesWithSchema(ctx, api, policy)
+
+ return policyLookupObservation{policy: policy, found: true, matches: matches}, err
+}
diff --git a/internal/provider/production_readiness_test.go b/internal/provider/production_readiness_test.go
new file mode 100644
index 0000000..9081ff3
--- /dev/null
+++ b/internal/provider/production_readiness_test.go
@@ -0,0 +1,440 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information regarding
+// copyright ownership. The ASF licenses this file to You under the Apache
+// License, Version 2.0 (the "License"); you may not use this file except in
+// compliance with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package provider
+
+import (
+ "context"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strconv"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/apache/terraform-provider-paimon/internal/client"
+ "github.com/hashicorp/terraform-plugin-framework/attr"
+ "github.com/hashicorp/terraform-plugin-framework/diag"
+ "github.com/hashicorp/terraform-plugin-framework/resource"
+ "github.com/hashicorp/terraform-plugin-framework/tfsdk"
+ "github.com/hashicorp/terraform-plugin-framework/types"
+ "github.com/stretchr/testify/require"
+)
+
+func TestDatabaseFailedOptionRemovalRetainsManagement(t *testing.T) {
+ ctx := context.Background()
+ remote := &client.Database{Name: "analytics", Options: map[string]string{"owner": "old"}}
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeProviderJSON(t, w, client.ConfigResponse{Defaults: map[string]string{"prefix": "catalog"}})
+
+ return
+ }
+ if r.Method == http.MethodPost {
+ w.WriteHeader(http.StatusServiceUnavailable)
+
+ return
+ }
+ writeProviderJSON(t, w, remote)
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL, RecoveryTimeout: 100 * time.Millisecond})
+ require.NoError(t, err)
+ res := &databaseResource{client: api}
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ var ds diag.Diagnostics
+ model := databaseResourceModel{Options: types.MapValueMust(types.StringType, map[string]attr.Value{"owner": types.StringValue("old")})}
+ setDatabaseResourceModel(ctx, &model, remote, &ds)
+ require.False(t, ds.HasError(), ds)
+ state := tfsdk.State{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, &model).HasError())
+ model.Options = types.MapValueMust(types.StringType, map[string]attr.Value{})
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, &model).HasError())
+ resp := resource.UpdateResponse{State: state}
+ res.Update(ctx, resource.UpdateRequest{State: state, Plan: plan}, &resp)
+ require.True(t, resp.Diagnostics.HasError(), resp.Diagnostics)
+ require.Empty(t, resp.Diagnostics.Warnings())
+ var after databaseResourceModel
+ require.False(t, resp.State.Get(ctx, &after).HasError())
+ require.Len(t, after.Options.Elements(), 1)
+ require.Equal(t, "old", remote.Options["owner"])
+}
+
+func TestTableFailedOptionRemovalRetainsManagement(t *testing.T) {
+ ctx := context.Background()
+ remote := &client.Table{Database: "analytics", Name: "events", Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT"}}, Options: map[string]string{"retention": "old"}}}
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeProviderJSON(t, w, client.ConfigResponse{Defaults: map[string]string{"prefix": "catalog"}})
+
+ return
+ }
+ if r.Method == http.MethodPost {
+ w.WriteHeader(http.StatusServiceUnavailable)
+
+ return
+ }
+ writeProviderJSON(t, w, remote)
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL, RecoveryTimeout: 100 * time.Millisecond})
+ require.NoError(t, err)
+ res := &tableResource{client: api}
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ var ds diag.Diagnostics
+ model := tableResourceModel{Fields: types.ListNull(types.ObjectType{AttrTypes: tableFieldAttrTypes()}), Options: types.MapValueMust(types.StringType, map[string]attr.Value{"retention": types.StringValue("old")})}
+ setTableResourceModel(ctx, &model, remote, &ds)
+ require.False(t, ds.HasError(), ds)
+ state := tfsdk.State{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, &model).HasError())
+ model.Options = types.MapValueMust(types.StringType, map[string]attr.Value{})
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, &model).HasError())
+ resp := resource.UpdateResponse{State: state}
+ res.Update(ctx, resource.UpdateRequest{State: state, Plan: plan}, &resp)
+ require.True(t, resp.Diagnostics.HasError(), resp.Diagnostics)
+ require.Empty(t, resp.Diagnostics.Warnings())
+ var after tableResourceModel
+ require.False(t, resp.State.Get(ctx, &after).HasError())
+ require.Len(t, after.Options.Elements(), 1)
+ require.Equal(t, "old", remote.Schema.Options["retention"])
+}
+
+func TestNotNullTypeRoundTripPreservesConfiguration(t *testing.T) {
+ ctx := context.Background()
+ var ds diag.Diagnostics
+ field := tableFieldForTest("id", types.Int64Value(0))
+ field.Type = types.StringValue("BIGINT NOT NULL")
+ field.Nullable = types.BoolValue(false)
+ managed := fieldsValueFromModels(ctx, []tableFieldModel{field}, &ds)
+ remote := resourceFieldsValueFromRemote(ctx, managed, []client.Field{{ID: 0, Name: "id", Type: "BIGINT NOT NULL"}}, &ds)
+ var got []tableFieldModel
+ require.False(t, remote.ElementsAs(ctx, &got, false).HasError())
+ require.False(t, ds.HasError(), ds)
+ require.Equal(t, "BIGINT NOT NULL", got[0].Type.ValueString())
+}
+
+func TestImportedImmutableOptionCanBecomeManaged(t *testing.T) {
+ ctx := context.Background()
+ remote := &client.Table{Database: "analytics", Name: "events", Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT"}}, Options: map[string]string{"merge-engine": "partial-update"}}}
+ var ds diag.Diagnostics
+ imported := tableResourceModel{Fields: types.ListNull(types.ObjectType{AttrTypes: tableFieldAttrTypes()}), Options: types.MapNull(types.StringType)}
+ setTableResourceModel(ctx, &imported, remote, &ds)
+ require.False(t, ds.HasError(), ds)
+ require.True(t, imported.Options.IsNull())
+ plannedOptions := types.MapValueMust(types.StringType, map[string]attr.Value{"merge-engine": types.StringValue("partial-update")})
+ require.Equal(t, plannedOptions, imported.ServerOptions)
+ baseline := effectiveManagedTableOptions(imported.Options, plannedOptions, imported.ServerOptions)
+ require.False(t, immutableTableOptionsChanged(baseline, plannedOptions))
+}
+
+func TestPolicyCreateRejectsExistingPolicy(t *testing.T) {
+ for _, mask := range []bool{false, true} {
+ for _, status := range []int{http.StatusConflict, http.StatusForbidden} {
+ t.Run(fmt.Sprintf("mask=%t/status=%d", mask, status), func(t *testing.T) {
+ ctx := context.Background()
+ reads, mutations := 0, 0
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeProviderJSON(t, w, client.ConfigResponse{Defaults: map[string]string{"prefix": "catalog"}})
+
+ return
+ }
+ if r.Method == http.MethodPost {
+ mutations++
+ w.WriteHeader(status)
+
+ return
+ }
+ reads++
+ writeProviderJSON(t, w, client.ListPoliciesResponse{Policies: []client.DataPolicy{{
+ Resource: client.PermissionResource{Type: client.ResourceTypeTable, Database: "db", Table: "tbl"},
+ Principal: "role:reader", RowFilter: &client.RowFilter{Predicate: `{"kind":"LEAF"}`},
+ ColumnMask: &client.ColumnMask{OnColumn: "secret", Transform: `{"name":"CONCAT"}`},
+ }}})
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL, RecoveryTimeout: 100 * time.Millisecond})
+ require.NoError(t, err)
+ var res resource.Resource
+ var model any
+ if mask {
+ res = &columnMaskResource{client: api}
+ model = &columnMaskResourceModel{Database: types.StringValue("db"), Table: types.StringValue("tbl"), Principal: types.StringValue("role:reader"), Column: types.StringValue("secret"), Transform: types.StringValue(`{"name":"CONCAT"}`)}
+ } else {
+ res = &rowFilterResource{client: api}
+ model = &rowFilterResourceModel{Database: types.StringValue("db"), Table: types.StringValue("tbl"), Principal: types.StringValue("role:reader"), Predicate: types.StringValue(`{"kind":"LEAF"}`)}
+ }
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, model).HasError())
+ resp := resource.CreateResponse{State: tfsdk.State{Schema: sr.Schema}}
+ res.Create(ctx, resource.CreateRequest{Plan: plan}, &resp)
+ require.True(t, resp.Diagnostics.HasError())
+ require.True(t, resp.State.Raw.IsNull(), "a rejected create must not establish ownership")
+ require.Zero(t, reads, "definitive rejections must not enter adoption reconciliation")
+ require.Equal(t, 1, mutations)
+ })
+ }
+ }
+}
+
+func TestPolicyUpdateRequiresMaintenanceOptIn(t *testing.T) {
+ ctx := context.Background()
+ for _, mask := range []bool{false, true} {
+ t.Run(strconv.FormatBool(mask), func(t *testing.T) {
+ requests := 0
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ requests++
+ w.WriteHeader(http.StatusInternalServerError)
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL, RecoveryTimeout: 100 * time.Millisecond})
+ require.NoError(t, err)
+ var res resource.Resource
+ var before, after any
+ if mask {
+ res = &columnMaskResource{client: api}
+ m := columnMaskResourceModel{ID: types.StringValue("existing"), Database: types.StringValue("db"), Table: types.StringValue("tbl"), Principal: types.StringValue("role:reader"), Column: types.StringValue("secret"), Transform: types.StringValue(`{"name":"old"}`)}
+ n := m
+ n.Transform = types.StringValue(`{"name":"new"}`)
+ before, after = &m, &n
+ } else {
+ res = &rowFilterResource{client: api}
+ m := rowFilterResourceModel{ID: types.StringValue("existing"), Database: types.StringValue("db"), Table: types.StringValue("tbl"), Principal: types.StringValue("role:reader"), Predicate: types.StringValue(`{"kind":"old"}`)}
+ n := m
+ n.Predicate = types.StringValue(`{"kind":"new"}`)
+ before, after = &m, &n
+ }
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ state := tfsdk.State{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, before).HasError())
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, after).HasError())
+ planned := resource.ModifyPlanResponse{Plan: plan}
+ res.(resource.ResourceWithModifyPlan).ModifyPlan(ctx, resource.ModifyPlanRequest{State: state, Plan: plan}, &planned)
+ require.True(t, planned.Diagnostics.HasError())
+ response := resource.UpdateResponse{State: state}
+ res.Update(ctx, resource.UpdateRequest{State: state, Plan: plan}, &response)
+ require.True(t, response.Diagnostics.HasError())
+ require.True(t, response.State.Raw.Equal(state.Raw))
+ require.Zero(t, requests, "blocked policy update must not remove protection")
+ })
+ }
+}
+
+func TestNotNullTypeDriftRemainsReadable(t *testing.T) {
+ ctx := context.Background()
+ var diags diag.Diagnostics
+ field := tableFieldForTest("id", types.Int64Value(0))
+ field.Type = types.StringValue("BIGINT NOT NULL")
+ field.Nullable = types.BoolValue(false)
+ managed := fieldsValueFromModels(ctx, []tableFieldModel{field}, &diags)
+ observed := resourceFieldsValueFromRemote(ctx, managed, []client.Field{{ID: 0, Name: "id", Type: "BIGINT"}}, &diags)
+ var fields []tableFieldModel
+ require.False(t, observed.ElementsAs(ctx, &fields, false).HasError())
+ require.Equal(t, "BIGINT", fields[0].Type.ValueString())
+ require.True(t, fields[0].Nullable.ValueBool())
+ schemaFromResourceModel(ctx, &tableResourceModel{Fields: observed, Options: types.MapNull(types.StringType)}, &diags)
+ require.False(t, diags.HasError(), diags)
+}
+
+func TestTableUpdateRejectsResolvedNewFieldIDBeforeMutation(t *testing.T) {
+ ctx := context.Background()
+ var diags diag.Diagnostics
+ remote := &client.Table{Database: "analytics", Name: "events", Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT"}}}}
+ model := tableResourceModel{Fields: types.ListNull(types.ObjectType{AttrTypes: tableFieldAttrTypes()}), Options: types.MapNull(types.StringType)}
+ setTableResourceModel(ctx, &model, remote, &diags)
+ require.False(t, diags.HasError(), diags)
+ res := &tableResource{} // No client: validation must finish before any REST call.
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ state := tfsdk.State{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, &model).HasError())
+ fields := fieldModelsFromRemote(remote.Schema.Fields)
+ fields = append(fields, tableFieldForTest("new_value", types.Int64Value(100)))
+ model.Fields = fieldsValueFromModels(ctx, fields, &diags)
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, &model).HasError())
+ response := resource.UpdateResponse{State: state}
+ res.Update(ctx, resource.UpdateRequest{State: state, Plan: plan}, &response)
+ require.True(t, response.Diagnostics.HasError())
+ require.Contains(t, response.Diagnostics.Errors()[0].Summary(), "New field IDs")
+ require.True(t, response.State.Raw.Equal(state.Raw))
+}
+
+func TestOptionReconciliationRequiresPresenceAndRemoval(t *testing.T) {
+ require.False(t, optionsConverged(map[string]string{}, map[string]string{"owner": ""}, nil))
+ require.True(t, optionsConverged(map[string]string{"owner": ""}, map[string]string{"owner": ""}, nil))
+ require.False(t, optionsConverged(map[string]string{"owner": "old"}, nil, []string{"owner"}))
+ require.True(t, optionsConverged(map[string]string{"server-only": "kept"}, nil, []string{"owner"}))
+}
+
+func TestTableReplacementRequiresOptIn(t *testing.T) {
+ ctx := context.Background()
+ for _, changed := range []string{"name", "database", "primary_keys", "partition_keys"} {
+ for _, allowed := range []bool{false, true} {
+ t.Run(fmt.Sprintf("%s/allowed=%t", changed, allowed), func(t *testing.T) {
+ res := &tableResource{}
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ var diags diag.Diagnostics
+ before := tableResourceModel{Fields: types.ListNull(types.ObjectType{AttrTypes: tableFieldAttrTypes()}), Options: types.MapNull(types.StringType)}
+ remote := &client.Table{Database: "analytics", Name: "events", Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT NOT NULL"}}}}
+ setTableResourceModel(ctx, &before, remote, &diags)
+ require.False(t, diags.HasError(), diags)
+ state := tfsdk.State{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, &before).HasError())
+ after := before
+ after.AllowReplacement = types.BoolValue(allowed)
+ switch changed {
+ case "name":
+ after.Name = types.StringValue("renamed")
+ case "database":
+ after.Database = types.StringValue("other")
+ case "primary_keys":
+ after.Options = types.MapValueMust(types.StringType, map[string]attr.Value{"primary-key": types.StringValue("id")})
+ case "partition_keys":
+ after.PartitionKeys = types.ListValueMust(types.StringType, []attr.Value{types.StringValue("id")})
+ }
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, &after).HasError())
+ response := resource.ModifyPlanResponse{Plan: plan}
+ res.ModifyPlan(ctx, resource.ModifyPlanRequest{State: state, Plan: plan, Config: tfsdk.Config{Schema: sr.Schema, Raw: plan.Raw}}, &response)
+ if allowed {
+ require.False(t, response.Diagnostics.HasError(), response.Diagnostics)
+ require.Len(t, response.RequiresReplace, 1)
+ } else {
+ require.True(t, response.Diagnostics.HasError())
+ require.Contains(t, response.Diagnostics.Errors()[0].Summary(), "Destructive table change")
+ }
+ })
+ }
+ }
+}
+
+func TestDatabaseSuccessfulButStaleRemovalRetainsManagement(t *testing.T) {
+ ctx := context.Background()
+ var visible atomic.Bool
+ remote := &client.Database{Name: "analytics", Options: map[string]string{"owner": "old"}}
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeProviderJSON(t, w, client.ConfigResponse{Defaults: map[string]string{"prefix": "catalog"}})
+
+ return
+ }
+ if r.Method == http.MethodPost {
+ w.WriteHeader(http.StatusOK)
+
+ return
+ }
+ if visible.Load() {
+ writeProviderJSON(t, w, &client.Database{Name: remote.Name, Options: map[string]string{}})
+
+ return
+ }
+ writeProviderJSON(t, w, remote)
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL, RecoveryTimeout: 100 * time.Millisecond})
+ require.NoError(t, err)
+ res := &databaseResource{client: api}
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ var ds diag.Diagnostics
+ model := databaseResourceModel{Options: types.MapValueMust(types.StringType, map[string]attr.Value{"owner": types.StringValue("old")})}
+ setDatabaseResourceModel(ctx, &model, remote, &ds)
+ require.False(t, ds.HasError(), ds)
+ state := tfsdk.State{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, &model).HasError())
+ model.Options = types.MapValueMust(types.StringType, map[string]attr.Value{})
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, &model).HasError())
+ resp := resource.UpdateResponse{State: state}
+ res.Update(ctx, resource.UpdateRequest{State: state, Plan: plan}, &resp)
+ require.True(t, resp.Diagnostics.HasError(), resp.Diagnostics)
+ require.Empty(t, resp.Diagnostics.Warnings())
+ var after databaseResourceModel
+ require.False(t, resp.State.Get(ctx, &after).HasError())
+ require.Len(t, after.Options.Elements(), 1)
+ require.Equal(t, "old", remote.Options["owner"])
+
+ visible.Store(true)
+ read := resource.ReadResponse{State: resp.State}
+ res.Read(ctx, resource.ReadRequest{State: resp.State}, &read)
+ require.False(t, read.Diagnostics.HasError(), read.Diagnostics)
+ require.False(t, read.State.Get(ctx, &after).HasError())
+ require.Empty(t, after.Options.Elements(), "a subsequent refresh must observe the completed removal")
+}
+
+func TestTableSuccessfulButStaleRemovalRetainsManagement(t *testing.T) {
+ ctx := context.Background()
+ var visible atomic.Bool
+ remote := &client.Table{Database: "analytics", Name: "events", Schema: client.Schema{Fields: []client.Field{{ID: 0, Name: "id", Type: "BIGINT"}}, Options: map[string]string{"retention": "old"}}}
+ server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeProviderJSON(t, w, client.ConfigResponse{Defaults: map[string]string{"prefix": "catalog"}})
+
+ return
+ }
+ if r.Method == http.MethodPost {
+ w.WriteHeader(http.StatusOK)
+
+ return
+ }
+ if visible.Load() {
+ writeProviderJSON(t, w, &client.Table{Database: remote.Database, Name: remote.Name, Schema: client.Schema{Fields: remote.Schema.Fields, Options: map[string]string{}}})
+
+ return
+ }
+ writeProviderJSON(t, w, remote)
+ }))
+ defer server.Close()
+ api, err := client.New(client.Config{URI: server.URL, RecoveryTimeout: 100 * time.Millisecond})
+ require.NoError(t, err)
+ res := &tableResource{client: api}
+ var sr resource.SchemaResponse
+ res.Schema(ctx, resource.SchemaRequest{}, &sr)
+ var ds diag.Diagnostics
+ model := tableResourceModel{Fields: types.ListNull(types.ObjectType{AttrTypes: tableFieldAttrTypes()}), Options: types.MapValueMust(types.StringType, map[string]attr.Value{"retention": types.StringValue("old")})}
+ setTableResourceModel(ctx, &model, remote, &ds)
+ require.False(t, ds.HasError(), ds)
+ state := tfsdk.State{Schema: sr.Schema}
+ require.False(t, state.Set(ctx, &model).HasError())
+ model.Options = types.MapValueMust(types.StringType, map[string]attr.Value{})
+ plan := tfsdk.Plan{Schema: sr.Schema}
+ require.False(t, plan.Set(ctx, &model).HasError())
+ resp := resource.UpdateResponse{State: state}
+ res.Update(ctx, resource.UpdateRequest{State: state, Plan: plan}, &resp)
+ require.True(t, resp.Diagnostics.HasError(), resp.Diagnostics)
+ require.Empty(t, resp.Diagnostics.Warnings())
+ var after tableResourceModel
+ require.False(t, resp.State.Get(ctx, &after).HasError())
+ require.Len(t, after.Options.Elements(), 1)
+ require.Equal(t, "old", remote.Schema.Options["retention"])
+
+ visible.Store(true)
+ read := resource.ReadResponse{State: resp.State}
+ res.Read(ctx, resource.ReadRequest{State: resp.State}, &read)
+ require.False(t, read.Diagnostics.HasError(), read.Diagnostics)
+ require.False(t, read.State.Get(ctx, &after).HasError())
+ require.Empty(t, after.Options.Elements(), "a subsequent refresh must observe the completed removal")
+}
diff --git a/internal/provider/provider.go b/internal/provider/provider.go
index cba8661..ec0c006 100644
--- a/internal/provider/provider.go
+++ b/internal/provider/provider.go
@@ -20,10 +20,13 @@ package provider
import (
"context"
"fmt"
+ "time"
"github.com/apache/terraform-provider-paimon/internal/client"
+ "github.com/hashicorp/terraform-plugin-framework-validators/int64validator"
"github.com/hashicorp/terraform-plugin-framework-validators/stringvalidator"
"github.com/hashicorp/terraform-plugin-framework/datasource"
+ "github.com/hashicorp/terraform-plugin-framework/diag"
"github.com/hashicorp/terraform-plugin-framework/provider"
"github.com/hashicorp/terraform-plugin-framework/provider/schema"
"github.com/hashicorp/terraform-plugin-framework/resource"
@@ -31,28 +34,53 @@ import (
"github.com/hashicorp/terraform-plugin-framework/types"
)
-var _ provider.Provider = &paimonProvider{}
+var (
+ _ provider.Provider = &paimonProvider{}
+ _ provider.ProviderWithValidateConfig = &paimonProvider{}
+)
+
+func (p *paimonProvider) ValidateConfig(ctx context.Context, req provider.ValidateConfigRequest, resp *provider.ValidateConfigResponse) {
+ var config paimonProviderModel
+ resp.Diagnostics.Append(req.Config.Get(ctx, &config)...)
+ if !resp.Diagnostics.HasError() {
+ validateAuthenticationConfig(config, &resp.Diagnostics)
+ }
+}
+
+func validateAuthenticationConfig(config paimonProviderModel, diags *diag.Diagnostics) {
+ if config.TokenProvider.IsUnknown() {
+ return
+ }
+ if knownString(config.TokenProvider) == client.AuthProviderBearer && hasDLFConfiguration(config) {
+ diags.AddError("Conflicting authentication configuration", "token_provider = bear cannot be combined with dlf_* configuration. Configure one authentication method.")
+ }
+ if knownString(config.Token) != "" && (knownString(config.TokenProvider) == client.AuthProviderDLF || hasDLFConfiguration(config)) {
+ diags.AddError("Conflicting authentication configuration", "A bearer token cannot be combined with DLF authentication or dlf_* configuration.")
+ }
+}
type paimonProvider struct {
version string
}
type paimonProviderModel struct {
- URI types.String `tfsdk:"uri"`
- Warehouse types.String `tfsdk:"warehouse"`
- TokenProvider types.String `tfsdk:"token_provider"`
- Token types.String `tfsdk:"token"`
- DLFRegion types.String `tfsdk:"dlf_region"`
- DLFSigningAlgorithm types.String `tfsdk:"dlf_signing_algorithm"`
- DLFAccessKeyID types.String `tfsdk:"dlf_access_key_id"`
- DLFAccessKeySecret types.String `tfsdk:"dlf_access_key_secret"`
- DLFSecurityToken types.String `tfsdk:"dlf_security_token"`
- DLFTokenPath types.String `tfsdk:"dlf_token_path"`
- DLFTokenLoader types.String `tfsdk:"dlf_token_loader"`
- DLFECSMetadataURL types.String `tfsdk:"dlf_ecs_metadata_url"`
- DLFECSRoleName types.String `tfsdk:"dlf_ecs_role_name"`
- Prefix types.String `tfsdk:"prefix"`
- Headers types.Map `tfsdk:"headers"`
+ URI types.String `tfsdk:"uri"`
+ Warehouse types.String `tfsdk:"warehouse"`
+ TokenProvider types.String `tfsdk:"token_provider"`
+ Token types.String `tfsdk:"token"`
+ DLFRegion types.String `tfsdk:"dlf_region"`
+ DLFSigningAlgorithm types.String `tfsdk:"dlf_signing_algorithm"`
+ DLFAccessKeyID types.String `tfsdk:"dlf_access_key_id"`
+ DLFAccessKeySecret types.String `tfsdk:"dlf_access_key_secret"`
+ DLFSecurityToken types.String `tfsdk:"dlf_security_token"`
+ DLFTokenPath types.String `tfsdk:"dlf_token_path"`
+ DLFTokenLoader types.String `tfsdk:"dlf_token_loader"`
+ DLFECSMetadataURL types.String `tfsdk:"dlf_ecs_metadata_url"`
+ DLFECSRoleName types.String `tfsdk:"dlf_ecs_role_name"`
+ Prefix types.String `tfsdk:"prefix"`
+ Headers types.Map `tfsdk:"headers"`
+ RequestTimeoutSeconds types.Int64 `tfsdk:"request_timeout_seconds"`
+ RecoveryTimeoutSeconds types.Int64 `tfsdk:"recovery_timeout_seconds"`
}
func New(version string) func() provider.Provider {
@@ -70,6 +98,16 @@ func (p *paimonProvider) Schema(_ context.Context, _ provider.SchemaRequest, res
resp.Schema = schema.Schema{
Description: "Use Terraform to manage databases, tables, permissions, and data policies through an Apache Paimon REST Catalog.",
Attributes: map[string]schema.Attribute{
+ "request_timeout_seconds": schema.Int64Attribute{
+ Description: "Timeout for each REST or metadata HTTP request, in seconds. Defaults to 30. Valid values: 1 to 3600.",
+ Optional: true,
+ Validators: []validator.Int64{int64validator.Between(1, 3600)},
+ },
+ "recovery_timeout_seconds": schema.Int64Attribute{
+ Description: "Maximum reconciliation time after a mutation, in seconds. Defaults to 5. Set this to cover the catalog's visibility delay. Valid values: 1 to 3600.",
+ Optional: true,
+ Validators: []validator.Int64{int64validator.Between(1, 3600)},
+ },
"uri": schema.StringAttribute{
Description: "Base URI of the Paimon REST Catalog server.",
Required: true,
@@ -146,6 +184,7 @@ func (p *paimonProvider) Schema(_ context.Context, _ provider.SchemaRequest, res
func (p *paimonProvider) Configure(ctx context.Context, req provider.ConfigureRequest, resp *provider.ConfigureResponse) {
var data paimonProviderModel
resp.Diagnostics.Append(req.Config.Get(ctx, &data)...)
+ validateAuthenticationConfig(data, &resp.Diagnostics)
if resp.Diagnostics.HasError() {
return
}
@@ -192,13 +231,15 @@ func (p *paimonProvider) Configure(ctx context.Context, req provider.ConfigureRe
}
api, err := client.New(client.Config{
- URI: data.URI.ValueString(),
- Warehouse: knownString(data.Warehouse),
- AuthProvider: authProvider,
- Token: knownString(data.Token),
- DLF: dlfConfig,
- Prefix: knownString(data.Prefix),
- Headers: headers,
+ RequestTimeout: time.Duration(data.RequestTimeoutSeconds.ValueInt64()) * time.Second,
+ RecoveryTimeout: time.Duration(data.RecoveryTimeoutSeconds.ValueInt64()) * time.Second,
+ URI: data.URI.ValueString(),
+ Warehouse: knownString(data.Warehouse),
+ AuthProvider: authProvider,
+ Token: knownString(data.Token),
+ DLF: dlfConfig,
+ Prefix: knownString(data.Prefix),
+ Headers: headers,
})
if err != nil {
resp.Diagnostics.AddError("Invalid Paimon provider configuration", fmt.Sprintf("Unable to create the REST Catalog client: %s", err))
diff --git a/internal/provider/provider_management_test.go b/internal/provider/provider_management_test.go
index eabb5d6..a5fa835 100644
--- a/internal/provider/provider_management_test.go
+++ b/internal/provider/provider_management_test.go
@@ -188,11 +188,12 @@ func TestRowFilterResourceLifecyclePreservesEquivalentJSON(t *testing.T) {
require.False(t, schemaResponse.Diagnostics.HasError(), schemaResponse.Diagnostics.Errors())
configuredPredicate := `{ "op": "eq", "field": "tenant_id" }`
planModel := rowFilterResourceModel{
- ID: types.StringUnknown(),
- Database: types.StringValue("analytics"),
- Table: types.StringValue("events"),
- Principal: types.StringValue("role:analyst"),
- Predicate: types.StringValue(configuredPredicate),
+ AllowNonAtomicUpdate: types.BoolValue(true),
+ ID: types.StringUnknown(),
+ Database: types.StringValue("analytics"),
+ Table: types.StringValue("events"),
+ Principal: types.StringValue("role:analyst"),
+ Predicate: types.StringValue(configuredPredicate),
}
plan := tfsdk.Plan{Schema: schemaResponse.Schema}
require.False(t, plan.Set(ctx, &planModel).HasError())
@@ -272,12 +273,13 @@ func TestColumnMaskResourceLifecycle(t *testing.T) {
var schemaResponse resource.SchemaResponse
managed.Schema(ctx, resource.SchemaRequest{}, &schemaResponse)
planModel := columnMaskResourceModel{
- ID: types.StringUnknown(),
- Database: types.StringValue("analytics"),
- Table: types.StringValue("events"),
- Principal: types.StringValue("role:analyst"),
- Column: types.StringValue("email"),
- Transform: types.StringValue(`{"type":"null"}`),
+ AllowNonAtomicUpdate: types.BoolValue(true),
+ ID: types.StringUnknown(),
+ Database: types.StringValue("analytics"),
+ Table: types.StringValue("events"),
+ Principal: types.StringValue("role:analyst"),
+ Column: types.StringValue("email"),
+ Transform: types.StringValue(`{"type":"null"}`),
}
plan := tfsdk.Plan{Schema: schemaResponse.Schema}
require.False(t, plan.Set(ctx, &planModel).HasError())
@@ -337,11 +339,12 @@ func TestRowFilterUpdateRestoresPreviousPolicyWhenReplacementFails(t *testing.T)
var schemaResponse resource.SchemaResponse
managed.Schema(ctx, resource.SchemaRequest{}, &schemaResponse)
stateModel := rowFilterResourceModel{
- ID: types.StringValue("database=analytics&principal=role%3Aanalyst&table=events"),
- Database: types.StringValue("analytics"),
- Table: types.StringValue("events"),
- Principal: types.StringValue("role:analyst"),
- Predicate: types.StringValue(`{"version":"old"}`),
+ AllowNonAtomicUpdate: types.BoolValue(true),
+ ID: types.StringValue("database=analytics&principal=role%3Aanalyst&table=events"),
+ Database: types.StringValue("analytics"),
+ Table: types.StringValue("events"),
+ Principal: types.StringValue("role:analyst"),
+ Predicate: types.StringValue(`{"version":"old"}`),
}
planModel := stateModel
planModel.Predicate = types.StringValue(`{"version":"new"}`)
@@ -789,11 +792,13 @@ func TestPolicyCreateReconcilesLostResponse(t *testing.T) {
api, err := client.New(client.Config{URI: server.URL})
require.NoError(t, err)
- result := createPolicyWithReconciliation(context.Background(), api, test.spec)
+ result := createPolicyWithReconciliation(context.Background(), api, test.spec, false)
require.True(t, result.accepted)
require.NoError(t, result.err)
require.NotNil(t, result.observed)
- assert.True(t, test.spec.matches(*result.observed))
+ matches, matchErr := test.spec.matchesWithSchema(context.Background(), api, *result.observed)
+ require.NoError(t, matchErr)
+ assert.True(t, matches)
assert.NotEmpty(t, result.warning)
})
}
@@ -867,7 +872,9 @@ func TestPolicyReplacementRecoversCanceledDrop(t *testing.T) {
require.NoError(t, result.err)
mu.Lock()
require.NotNil(t, remote)
- assert.True(t, test.desired.matches(*remote))
+ matches, matchErr := test.desired.matchesWithSchema(context.Background(), api, *remote)
+ require.NoError(t, matchErr)
+ assert.True(t, matches)
mu.Unlock()
})
}
diff --git a/internal/provider/provider_test.go b/internal/provider/provider_test.go
index 066c0eb..4c0a899 100644
--- a/internal/provider/provider_test.go
+++ b/internal/provider/provider_test.go
@@ -203,7 +203,7 @@ func TestReservedTableOptionsValidator(t *testing.T) {
ConfigValue: options,
}, &response)
require.True(t, response.Diagnostics.HasError())
- assert.Contains(t, response.Diagnostics.Errors()[0].Detail(), "partition, primary-key")
+ assert.Contains(t, response.Diagnostics.Errors()[0].Detail(), "Do not configure partition in options")
}
func TestImmutableTableOptionsChanged(t *testing.T) {
@@ -401,7 +401,7 @@ func TestTableResourceLifecycle(t *testing.T) {
require.False(t, diagnostics.HasError(), diagnostics.Errors())
planModel := tableResourceModel{
ID: types.StringUnknown(),
- CatalogID: types.StringUnknown(),
+ ServerID: types.StringUnknown(),
Database: types.StringValue("analytics"),
Name: types.StringValue("events"),
Fields: fields,
@@ -484,7 +484,7 @@ func TestDatabaseCreateRetainsIdentityWhenReadBackFails(t *testing.T) {
planModel := databaseResourceModel{
ID: types.StringUnknown(),
- CatalogID: types.StringUnknown(),
+ ServerID: types.StringUnknown(),
Name: types.StringValue("analytics"),
Options: types.MapNull(types.StringType),
ServerOptions: types.MapUnknown(types.StringType),
diff --git a/internal/provider/real_catalog_acceptance_test.go b/internal/provider/real_catalog_acceptance_test.go
new file mode 100644
index 0000000..e3f365f
--- /dev/null
+++ b/internal/provider/real_catalog_acceptance_test.go
@@ -0,0 +1,95 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information regarding
+// copyright ownership. The ASF licenses this file to You under the Apache
+// License, Version 2.0 (the "License"); you may not use this file except in
+// compliance with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package provider
+
+import (
+ "crypto/rand"
+ "fmt"
+ "os"
+ "strings"
+ "testing"
+
+ "github.com/hashicorp/terraform-plugin-testing/helper/resource"
+ "github.com/hashicorp/terraform-plugin-testing/plancheck"
+ "github.com/stretchr/testify/require"
+)
+
+// This suite uses a supplied, isolated REST service. It never substitutes the
+// in-process acceptance fixture for a real-server compatibility result.
+func TestAccRealCatalog(t *testing.T) {
+ uri := os.Getenv("PAIMON_ACC_URI")
+ if os.Getenv("TF_ACC") == "" || uri == "" {
+ t.Skip("set TF_ACC=1 and PAIMON_ACC_URI to run live catalog acceptance")
+ }
+ require.Equal(t, "1", os.Getenv("PAIMON_ACC_ALLOW_MUTATIONS"), "live acceptance creates, alters and deletes its uniquely named database; set PAIMON_ACC_ALLOW_MUTATIONS=1 for an isolated test catalog")
+ var suffix [8]byte
+ _, err := rand.Read(suffix[:])
+ require.NoError(t, err)
+ database := fmt.Sprintf("terraform_acc_%x", suffix)
+ t.Logf("live acceptance database: %s (retain this name for cleanup if the service is unavailable)", database)
+ providerConfig := realCatalogProviderConfig(uri)
+ initial := providerConfig + managementTableConfig(database)
+ updated := strings.Replace(initial, `{ name = "secret", type = "STRING" }`, `{ name = "secret", type = "STRING" }, { name = "added", type = "STRING" }`, 1)
+ sources := `
+data "paimon_database" "read" { name = paimon_database.analytics.name }
+data "paimon_table" "read" { database = paimon_table.events.database
+ name = paimon_table.events.name
+}
+`
+ initial += sources
+ updated += sources
+ steps := []resource.TestStep{
+ {Config: initial, Check: resource.ComposeTestCheckFunc(
+ resource.TestCheckResourceAttr("data.paimon_database.read", "name", database),
+ resource.TestCheckResourceAttr("data.paimon_table.read", "name", "events"),
+ )},
+ {ResourceName: "paimon_table.events", ImportState: true, ImportStateVerify: true, ImportStateVerifyIgnore: []string{"options"}},
+ {ResourceName: "paimon_database.analytics", ImportState: true, ImportStateVerify: true, ImportStateVerifyIgnore: []string{"options"}},
+ {Config: updated, ConfigPlanChecks: resource.ConfigPlanChecks{PreApply: []plancheck.PlanCheck{plancheck.ExpectResourceAction("paimon_table.events", plancheck.ResourceActionUpdate)}}},
+ {Config: updated, PlanOnly: true},
+ }
+ if principal := os.Getenv("PAIMON_ACC_PRINCIPAL"); principal != "" {
+ // The caller provisions a test principal; the provider does not manage
+ // principals and must not grant rights to a production identity.
+ first := updated + managementResourceConfig(principal, false)
+ last := updated + managementResourceConfig(principal, true)
+ steps = append(steps,
+ resource.TestStep{Config: first},
+ resource.TestStep{ResourceName: "paimon_permission.read", ImportState: true, ImportStateVerify: true},
+ resource.TestStep{ResourceName: "paimon_row_filter.region", ImportState: true, ImportStateVerify: true, ImportStateVerifyIgnore: []string{"allow_non_atomic_update"}},
+ resource.TestStep{ResourceName: "paimon_column_mask.secret", ImportState: true, ImportStateVerify: true, ImportStateVerifyIgnore: []string{"allow_non_atomic_update"}},
+ resource.TestStep{Config: last},
+ resource.TestStep{Config: last, PlanOnly: true},
+ )
+ }
+ resource.Test(t, resource.TestCase{ProtoV6ProviderFactories: testAccProtoV6ProviderFactories, Steps: steps})
+}
+
+func realCatalogProviderConfig(uri string) string {
+ // Secrets stay in TF_VAR_* environment variables; they are never embedded
+ // into test configurations or failure messages.
+ attributes := []string{"token_provider", "token", "warehouse", "prefix", "dlf_region", "dlf_signing_algorithm", "dlf_access_key_id", "dlf_access_key_secret", "dlf_security_token", "dlf_token_loader", "dlf_token_path", "dlf_ecs_metadata_url", "dlf_ecs_role_name"}
+ var variables, configuration strings.Builder
+ fmt.Fprintf(&configuration, "provider \"paimon\" {\n uri = %q\n recovery_timeout_seconds = 60\n", uri)
+ for _, attribute := range attributes {
+ fmt.Fprintf(&variables, "variable \"paimon_%s\" {\n type = string\n default = null\n sensitive = true\n}\n", attribute)
+ fmt.Fprintf(&configuration, " %s = var.paimon_%s\n", attribute, attribute)
+ }
+ configuration.WriteString(" headers = var.paimon_headers\n}\n")
+ variables.WriteString("variable \"paimon_headers\" {\n type = map(string)\n default = {}\n sensitive = true\n}\n")
+
+ return variables.String() + configuration.String()
+}
diff --git a/internal/provider/resource_database.go b/internal/provider/resource_database.go
index c692ef2..676b4c3 100644
--- a/internal/provider/resource_database.go
+++ b/internal/provider/resource_database.go
@@ -42,7 +42,7 @@ type databaseResource struct {
type databaseResourceModel struct {
ID types.String `tfsdk:"id"`
- CatalogID types.String `tfsdk:"catalog_id"`
+ ServerID types.String `tfsdk:"server_id"`
Name types.String `tfsdk:"name"`
Options types.Map `tfsdk:"options"`
ServerOptions types.Map `tfsdk:"server_options"`
@@ -71,7 +71,7 @@ func (r *databaseResource) Schema(_ context.Context, _ resource.SchemaRequest, r
Computed: true,
PlanModifiers: []planmodifier.String{stringplanmodifier.UseStateForUnknown()},
},
- "catalog_id": schema.StringAttribute{Description: "Server-assigned database identifier.", Computed: true},
+ "server_id": schema.StringAttribute{Description: "Server-assigned database identifier.", Computed: true},
"name": schema.StringAttribute{
Description: "Database name.",
Required: true,
@@ -185,12 +185,18 @@ func (r *databaseResource) Update(ctx context.Context, req resource.UpdateReques
return
}
removals, updates := diffOptions(before, after)
+ stableCtx := context.WithoutCancel(ctx)
+ resp.Diagnostics.Append(resp.State.Set(stableCtx, &state)...)
+ if resp.Diagnostics.HasError() {
+ return
+ }
+ ready := func(observed *client.Database) bool {
+ return databaseMatchesOptions(observed, plan.Name.ValueString(), after, false) && optionsConverged(observed.Options, after, removals)
+ }
if len(removals) > 0 || len(updates) > 0 {
if err := r.client.AlterDatabase(ctx, plan.Name.ValueString(), removals, updates); err != nil {
if client.IsMutationOutcomeUncertain(err) {
- database, recoveryErr := r.lookupAfterMutation(ctx, plan.Name.ValueString(), func(observed *client.Database) bool {
- return databaseMatchesOptions(observed, plan.Name.ValueString(), after, false)
- })
+ database, recoveryErr := r.lookupAfterMutation(ctx, plan.Name.ValueString(), ready)
if recoveryErr == nil {
stableCtx := context.WithoutCancel(ctx)
setDatabaseResourceModel(stableCtx, &plan, database, &resp.Diagnostics)
@@ -205,17 +211,9 @@ func (r *databaseResource) Update(ctx context.Context, req resource.UpdateReques
return
}
}
- stableCtx := context.WithoutCancel(ctx)
- plan.ID = types.StringValue(plan.Name.ValueString())
- resp.Diagnostics.Append(resp.State.Set(stableCtx, &plan)...)
- if resp.Diagnostics.HasError() {
- return
- }
- database, err := r.lookupAfterMutation(ctx, plan.Name.ValueString(), func(observed *client.Database) bool {
- return databaseMatchesOptions(observed, plan.Name.ValueString(), after, false)
- })
+ database, err := r.lookupAfterMutation(ctx, plan.Name.ValueString(), ready)
if err != nil {
- resp.Diagnostics.AddError("Unable to verify Paimon database after update", err.Error()+". Terraform retained the planned database state.")
+ resp.Diagnostics.AddError("Unable to verify Paimon database after update", err.Error()+". Terraform retained the previous state so unverified removals remain managed.")
return
}
@@ -240,7 +238,7 @@ func (r *databaseResource) ImportState(ctx context.Context, req resource.ImportS
}
func (r *databaseResource) lookupAfterMutation(ctx context.Context, name string, ready func(*client.Database) bool) (*client.Database, error) {
- recoveryCtx, cancel := mutationRecoveryContext(ctx)
+ recoveryCtx, cancel := mutationRecoveryContext(ctx, r.client)
defer cancel()
database, found, converged, err := retryLookupUntil(recoveryCtx, func(attemptCtx context.Context) (*client.Database, bool, error) {
observed, lookupErr := r.client.GetDatabase(attemptCtx, name)
@@ -267,18 +265,13 @@ func databaseMatchesOptions(database *client.Database, name string, options map[
if database == nil || database.Name != name || exact && len(database.Options) != len(options) {
return false
}
- for key, value := range options {
- if database.Options[key] != value {
- return false
- }
- }
- return true
+ return optionsConverged(database.Options, options, nil)
}
func setDatabaseResourceModel(ctx context.Context, model *databaseResourceModel, database *client.Database, diags *diag.Diagnostics) {
model.ID = types.StringValue(database.Name)
- model.CatalogID = types.StringValue(database.ID)
+ model.ServerID = types.StringValue(database.ID)
model.Name = types.StringValue(database.Name)
model.Options = syncManagedOptions(ctx, model.Options, database.Options, diags)
model.ServerOptions = stringMapValue(ctx, database.Options, diags)
diff --git a/internal/provider/resource_permission.go b/internal/provider/resource_permission.go
index 6dcc1c0..eff1f79 100644
--- a/internal/provider/resource_permission.go
+++ b/internal/provider/resource_permission.go
@@ -346,7 +346,7 @@ func (r *permissionResource) readIntoState(ctx context.Context, model *permissio
}
func (r *permissionResource) readAfterMutation(ctx context.Context, model *permissionResourceModel, expected client.PermissionAssignment, diags *diag.Diagnostics) bool {
- recoveryCtx, cancel := mutationRecoveryContext(ctx)
+ recoveryCtx, cancel := mutationRecoveryContext(ctx, r.client)
defer cancel()
assignment, found, converged, err := retryLookupUntil(recoveryCtx, func(attemptCtx context.Context) (client.PermissionAssignment, bool, error) {
return r.lookup(attemptCtx, *model)
@@ -374,7 +374,7 @@ func (r *permissionResource) readAfterMutation(ctx context.Context, model *permi
}
func (r *permissionResource) reconcileFailedGrant(ctx context.Context, model permissionResourceModel, expected client.PermissionAssignment, grantErr error) (client.PermissionAssignment, bool, error) {
- recoveryCtx, cancel := mutationRecoveryContext(ctx)
+ recoveryCtx, cancel := mutationRecoveryContext(ctx, r.client)
defer cancel()
observed, found, converged, reconcileErr := retryLookupUntil(recoveryCtx, func(attemptCtx context.Context) (client.PermissionAssignment, bool, error) {
return r.lookup(attemptCtx, model)
diff --git a/internal/provider/resource_policy.go b/internal/provider/resource_policy.go
index d3b3b9c..d82b54c 100644
--- a/internal/provider/resource_policy.go
+++ b/internal/provider/resource_policy.go
@@ -29,6 +29,7 @@ import (
"github.com/hashicorp/terraform-plugin-framework/path"
"github.com/hashicorp/terraform-plugin-framework/resource"
"github.com/hashicorp/terraform-plugin-framework/resource/schema"
+ "github.com/hashicorp/terraform-plugin-framework/resource/schema/booldefault"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier"
"github.com/hashicorp/terraform-plugin-framework/schema/validator"
@@ -37,13 +38,81 @@ import (
const maxSerializedPolicyBytes = 60 * 1024
+const nonAtomicPolicyUpdateDetail = "The REST Catalog has no atomic policy update. Changing policy content drops and recreates the policy, leaving an interval without filtering or masking. Stop affected queries first, then explicitly set allow_non_atomic_update = true for the maintenance operation. Rollback cannot eliminate this interval."
+
+func nonAtomicPolicyUpdateAttribute() schema.BoolAttribute {
+ return schema.BoolAttribute{
+ Description: "Permit drop-and-create updates during a controlled maintenance window. Defaults to false because reads can be unfiltered or unmasked between requests.",
+ Optional: true,
+ Computed: true,
+ Default: booldefault.StaticBool(false),
+ }
+}
+
+func (r *rowFilterResource) ModifyPlan(ctx context.Context, req resource.ModifyPlanRequest, resp *resource.ModifyPlanResponse) {
+ validatePolicyUpdatePlan(ctx, r.client, req, resp, "predicate")
+}
+
+func (r *columnMaskResource) ModifyPlan(ctx context.Context, req resource.ModifyPlanRequest, resp *resource.ModifyPlanResponse) {
+ validatePolicyUpdatePlan(ctx, r.client, req, resp, "transform")
+}
+
+func validatePolicyUpdatePlan(ctx context.Context, api *client.Client, req resource.ModifyPlanRequest, resp *resource.ModifyPlanResponse, contentAttribute string) {
+ if req.State.Raw.IsNull() || req.Plan.Raw.IsNull() {
+ return
+ }
+ identities := []string{"database", "table", "principal"}
+ if contentAttribute == "transform" {
+ identities = append(identities, "column")
+ }
+ for _, name := range identities {
+ var previous, planned types.String
+ resp.Diagnostics.Append(req.State.GetAttribute(ctx, path.Root(name), &previous)...)
+ resp.Diagnostics.Append(req.Plan.GetAttribute(ctx, path.Root(name), &planned)...)
+ if resp.Diagnostics.HasError() || planned.IsUnknown() || !previous.Equal(planned) {
+ // Identity changes use replacement. Its destination table may not
+ // exist yet, and this guard only governs in-place content updates.
+ return
+ }
+ }
+ var before, after types.String
+ var allowed types.Bool
+ resp.Diagnostics.Append(req.State.GetAttribute(ctx, path.Root(contentAttribute), &before)...)
+ resp.Diagnostics.Append(req.Plan.GetAttribute(ctx, path.Root(contentAttribute), &after)...)
+ resp.Diagnostics.Append(req.Plan.GetAttribute(ctx, path.Root("allow_non_atomic_update"), &allowed)...)
+ if resp.Diagnostics.HasError() || before.IsUnknown() || after.IsUnknown() {
+ return
+ }
+ var database, table types.String
+ resp.Diagnostics.Append(req.Plan.GetAttribute(ctx, path.Root("database"), &database)...)
+ resp.Diagnostics.Append(req.Plan.GetAttribute(ctx, path.Root("table"), &table)...)
+ if resp.Diagnostics.HasError() || database.IsUnknown() || table.IsUnknown() {
+ return
+ }
+ policyType := client.PolicyTypeRowFilter
+ if contentAttribute == "transform" {
+ policyType = client.PolicyTypeColumnMasking
+ }
+ equal, err := equivalentPolicyContent(ctx, api, policySpec{database: database.ValueString(), table: table.ValueString(), policyType: policyType}, before.ValueString(), after.ValueString())
+ if err != nil {
+ resp.Diagnostics.AddError("Unable to compare Paimon policy", err.Error())
+
+ return
+ }
+ if !equal && !allowed.ValueBool() {
+ resp.Diagnostics.AddAttributeError(path.Root(contentAttribute), "Non-atomic policy update is disabled", nonAtomicPolicyUpdateDetail)
+ }
+}
+
var (
_ resource.Resource = &rowFilterResource{}
_ resource.ResourceWithImportState = &rowFilterResource{}
_ resource.ResourceWithValidateConfig = &rowFilterResource{}
+ _ resource.ResourceWithModifyPlan = &rowFilterResource{}
_ resource.Resource = &columnMaskResource{}
_ resource.ResourceWithImportState = &columnMaskResource{}
_ resource.ResourceWithValidateConfig = &columnMaskResource{}
+ _ resource.ResourceWithModifyPlan = &columnMaskResource{}
)
type rowFilterResource struct {
@@ -51,11 +120,12 @@ type rowFilterResource struct {
}
type rowFilterResourceModel struct {
- ID types.String `tfsdk:"id"`
- Database types.String `tfsdk:"database"`
- Table types.String `tfsdk:"table"`
- Principal types.String `tfsdk:"principal"`
- Predicate types.String `tfsdk:"predicate"`
+ ID types.String `tfsdk:"id"`
+ Database types.String `tfsdk:"database"`
+ Table types.String `tfsdk:"table"`
+ Principal types.String `tfsdk:"principal"`
+ Predicate types.String `tfsdk:"predicate"`
+ AllowNonAtomicUpdate types.Bool `tfsdk:"allow_non_atomic_update"`
}
type columnMaskResource struct {
@@ -63,12 +133,13 @@ type columnMaskResource struct {
}
type columnMaskResourceModel struct {
- ID types.String `tfsdk:"id"`
- Database types.String `tfsdk:"database"`
- Table types.String `tfsdk:"table"`
- Principal types.String `tfsdk:"principal"`
- Column types.String `tfsdk:"column"`
- Transform types.String `tfsdk:"transform"`
+ ID types.String `tfsdk:"id"`
+ Database types.String `tfsdk:"database"`
+ Table types.String `tfsdk:"table"`
+ Principal types.String `tfsdk:"principal"`
+ Column types.String `tfsdk:"column"`
+ Transform types.String `tfsdk:"transform"`
+ AllowNonAtomicUpdate types.Bool `tfsdk:"allow_non_atomic_update"`
}
func NewRowFilterResource() resource.Resource {
@@ -88,6 +159,7 @@ func (r *rowFilterResource) Schema(_ context.Context, _ resource.SchemaRequest,
resp.Schema = schema.Schema{
Description: "Manages one principal-scoped row-filter policy on a Paimon table. The table must enable query-auth.enabled. The management API is experimental in Paimon.",
Attributes: map[string]schema.Attribute{
+ "allow_non_atomic_update": nonAtomicPolicyUpdateAttribute(),
"id": schema.StringAttribute{
Description: "Stable URL-query identifier for the row-filter identity.",
Computed: true,
@@ -128,14 +200,14 @@ func (r *rowFilterResource) Create(ctx context.Context, req resource.CreateReque
if resp.Diagnostics.HasError() {
return
}
- result := createPolicyWithReconciliation(ctx, r.client, rowFilterSpec(plan))
+ result := createPolicyWithReconciliation(ctx, r.client, rowFilterSpec(plan), false)
if !result.accepted {
resp.Diagnostics.AddError("Unable to create Paimon row filter", result.err.Error())
return
}
plan.ID = types.StringValue(rowFilterID(plan))
- if result.observed != nil && result.observed.RowFilter != nil && !equivalentJSON(plan.Predicate.ValueString(), result.observed.RowFilter.Predicate) {
+ if result.err != nil && result.observed != nil && result.observed.RowFilter != nil && !equivalentJSON(plan.Predicate.ValueString(), result.observed.RowFilter.Predicate) {
plan.Predicate = types.StringValue(result.observed.RowFilter.Predicate)
}
stableCtx := context.WithoutCancel(ctx)
@@ -180,16 +252,27 @@ func (r *rowFilterResource) Update(ctx context.Context, req resource.UpdateReque
if resp.Diagnostics.HasError() {
return
}
- if equivalentJSON(state.Predicate.ValueString(), plan.Predicate.ValueString()) {
+ equal, err := equivalentPolicyContent(ctx, r.client, rowFilterSpec(plan), state.Predicate.ValueString(), plan.Predicate.ValueString())
+ if err != nil {
+ resp.Diagnostics.AddError("Unable to compare Paimon policy", err.Error())
+
+ return
+ }
+ if equal {
plan.ID = types.StringValue(rowFilterID(plan))
resp.Diagnostics.Append(resp.State.Set(stableCtx, &plan)...)
return
}
+ if !plan.AllowNonAtomicUpdate.ValueBool() {
+ resp.Diagnostics.AddError("Non-atomic policy update is disabled", nonAtomicPolicyUpdateDetail)
+
+ return
+ }
result := replacePolicyWithReconciliation(ctx, r.client, rowFilterSpec(state), rowFilterSpec(plan))
if result.desired {
plan.ID = types.StringValue(rowFilterID(plan))
- if result.observed != nil && result.observed.RowFilter != nil && !equivalentJSON(plan.Predicate.ValueString(), result.observed.RowFilter.Predicate) {
+ if result.err != nil && result.observed != nil && result.observed.RowFilter != nil && !equivalentJSON(plan.Predicate.ValueString(), result.observed.RowFilter.Predicate) {
plan.Predicate = types.StringValue(result.observed.RowFilter.Predicate)
}
resp.Diagnostics.Append(resp.State.Set(stableCtx, &plan)...)
@@ -242,7 +325,13 @@ func (r *rowFilterResource) readIntoState(ctx context.Context, model *rowFilterR
return false
}
remote := policy.RowFilter.Predicate
- if model.Predicate.IsNull() || model.Predicate.IsUnknown() || !equivalentJSON(model.Predicate.ValueString(), remote) {
+ equal, err := equivalentPolicyContent(ctx, r.client, rowFilterSpec(*model), model.Predicate.ValueString(), remote)
+ if err != nil {
+ diags.AddError("Unable to compare Paimon policy", err.Error())
+
+ return false
+ }
+ if model.Predicate.IsNull() || model.Predicate.IsUnknown() || !equal {
model.Predicate = types.StringValue(remote)
}
model.ID = types.StringValue(rowFilterID(*model))
@@ -259,6 +348,7 @@ func (r *columnMaskResource) Schema(_ context.Context, _ resource.SchemaRequest,
resp.Schema = schema.Schema{
Description: "Manages one principal-scoped column-masking policy on a Paimon table. The table must enable query-auth.enabled. The management API is experimental in Paimon.",
Attributes: map[string]schema.Attribute{
+ "allow_non_atomic_update": nonAtomicPolicyUpdateAttribute(),
"id": schema.StringAttribute{
Description: "Stable URL-query identifier for the column-mask identity.",
Computed: true,
@@ -300,14 +390,14 @@ func (r *columnMaskResource) Create(ctx context.Context, req resource.CreateRequ
if resp.Diagnostics.HasError() {
return
}
- result := createPolicyWithReconciliation(ctx, r.client, columnMaskSpec(plan))
+ result := createPolicyWithReconciliation(ctx, r.client, columnMaskSpec(plan), false)
if !result.accepted {
resp.Diagnostics.AddError("Unable to create Paimon column mask", result.err.Error())
return
}
plan.ID = types.StringValue(columnMaskID(plan))
- if result.observed != nil && result.observed.ColumnMask != nil && !equivalentJSON(plan.Transform.ValueString(), result.observed.ColumnMask.Transform) {
+ if result.err != nil && result.observed != nil && result.observed.ColumnMask != nil && !equivalentJSON(plan.Transform.ValueString(), result.observed.ColumnMask.Transform) {
plan.Transform = types.StringValue(result.observed.ColumnMask.Transform)
}
stableCtx := context.WithoutCancel(ctx)
@@ -352,16 +442,27 @@ func (r *columnMaskResource) Update(ctx context.Context, req resource.UpdateRequ
if resp.Diagnostics.HasError() {
return
}
- if equivalentJSON(state.Transform.ValueString(), plan.Transform.ValueString()) {
+ equal, err := equivalentPolicyContent(ctx, r.client, columnMaskSpec(plan), state.Transform.ValueString(), plan.Transform.ValueString())
+ if err != nil {
+ resp.Diagnostics.AddError("Unable to compare Paimon policy", err.Error())
+
+ return
+ }
+ if equal {
plan.ID = types.StringValue(columnMaskID(plan))
resp.Diagnostics.Append(resp.State.Set(stableCtx, &plan)...)
return
}
+ if !plan.AllowNonAtomicUpdate.ValueBool() {
+ resp.Diagnostics.AddError("Non-atomic policy update is disabled", nonAtomicPolicyUpdateDetail)
+
+ return
+ }
result := replacePolicyWithReconciliation(ctx, r.client, columnMaskSpec(state), columnMaskSpec(plan))
if result.desired {
plan.ID = types.StringValue(columnMaskID(plan))
- if result.observed != nil && result.observed.ColumnMask != nil && !equivalentJSON(plan.Transform.ValueString(), result.observed.ColumnMask.Transform) {
+ if result.err != nil && result.observed != nil && result.observed.ColumnMask != nil && !equivalentJSON(plan.Transform.ValueString(), result.observed.ColumnMask.Transform) {
plan.Transform = types.StringValue(result.observed.ColumnMask.Transform)
}
resp.Diagnostics.Append(resp.State.Set(stableCtx, &plan)...)
@@ -416,7 +517,13 @@ func (r *columnMaskResource) readIntoState(ctx context.Context, model *columnMas
return false
}
remote := policy.ColumnMask.Transform
- if model.Transform.IsNull() || model.Transform.IsUnknown() || !equivalentJSON(model.Transform.ValueString(), remote) {
+ equal, err := equivalentPolicyContent(ctx, r.client, columnMaskSpec(*model), model.Transform.ValueString(), remote)
+ if err != nil {
+ diags.AddError("Unable to compare Paimon policy", err.Error())
+
+ return false
+ }
+ if model.Transform.IsNull() || model.Transform.IsUnknown() || !equal {
model.Transform = types.StringValue(remote)
}
model.ID = types.StringValue(columnMaskID(*model))
@@ -466,17 +573,6 @@ func columnMaskSpec(model columnMaskResourceModel) policySpec {
}
}
-func (s policySpec) matches(policy client.DataPolicy) bool {
- switch s.policyType {
- case client.PolicyTypeRowFilter:
- return policy.RowFilter != nil && equivalentJSON(s.content, policy.RowFilter.Predicate)
- case client.PolicyTypeColumnMasking:
- return policy.ColumnMask != nil && policy.ColumnMask.OnColumn == s.column && equivalentJSON(s.content, policy.ColumnMask.Transform)
- default:
- return false
- }
-}
-
func lookupPolicy(ctx context.Context, api *client.Client, spec policySpec) (client.DataPolicy, bool, error) {
response, err := api.ListPolicies(ctx, client.ListPoliciesRequest{
Database: spec.database,
@@ -529,13 +625,19 @@ type policyCreateResult struct {
err error
}
-func createPolicyWithReconciliation(ctx context.Context, api *client.Client, spec policySpec) policyCreateResult {
+func createPolicyWithReconciliation(ctx context.Context, api *client.Client, spec policySpec, alreadyManaged bool) policyCreateResult {
createErr := api.CreatePolicy(ctx, spec.database, spec.table, spec.request)
- recoveryCtx, cancel := mutationRecoveryContext(ctx)
+ if createErr != nil && !alreadyManaged && !client.IsMutationOutcomeUncertain(createErr) {
+ return policyCreateResult{err: fmt.Errorf("creating the %s was rejected: %w. Import an existing policy explicitly; it has not been adopted into state", spec.label, createErr)}
+ }
+ recoveryCtx, cancel := mutationRecoveryContext(ctx, api)
defer cancel()
- observed, found, converged, reconcileErr := retryLookupUntil(recoveryCtx, func(attemptCtx context.Context) (client.DataPolicy, bool, error) {
- return lookupPolicy(attemptCtx, api, spec)
- }, spec.matches)
+ observation, found, converged, reconcileErr := retryLookupUntil(recoveryCtx, func(attemptCtx context.Context) (policyLookupObservation, bool, error) {
+ observation, err := observePolicy(attemptCtx, api, spec)
+
+ return observation, observation.found, err
+ }, func(observation policyLookupObservation) bool { return observation.matches })
+ observed := observation.policy
if createErr == nil {
result := policyCreateResult{accepted: true}
if reconcileErr != nil {
@@ -580,39 +682,41 @@ type policyReplacementResult struct {
}
type policyLookupObservation struct {
- policy client.DataPolicy
- found bool
+ policy client.DataPolicy
+ found bool
+ matches bool
}
func replacePolicyWithReconciliation(ctx context.Context, api *client.Client, previous, desired policySpec) policyReplacementResult {
mutationCtx := ctx
dropErr := api.DropPolicy(ctx, desired.database, desired.table, desired.policyType, desired.principal, desired.column)
if dropErr != nil && !client.IsNotFound(dropErr) {
- recoveryCtx, cancel := mutationRecoveryContext(ctx)
+ recoveryCtx, cancel := mutationRecoveryContext(ctx, api)
defer cancel()
observation, _, _, reconcileErr := retryLookupUntil(recoveryCtx, func(attemptCtx context.Context) (policyLookupObservation, bool, error) {
- observed, found, lookupErr := lookupPolicy(attemptCtx, api, desired)
- if lookupErr != nil {
- return policyLookupObservation{}, false, lookupErr
- }
+ observation, err := observePolicy(attemptCtx, api, desired)
- return policyLookupObservation{policy: observed, found: found}, true, nil
+ return observation, true, err
}, func(observed policyLookupObservation) bool {
- return !observed.found || desired.matches(observed.policy)
+ return !observed.found || observed.matches
})
if reconcileErr != nil {
return policyReplacementResult{err: fmt.Errorf("dropping the previous %s returned an error (%s), and bounded reconciliation could not establish the remote state: %w", previous.label, dropErr, reconcileErr)}
}
if observation.found {
observed := observation.policy
+ previousMatches, err := previous.matchesWithSchema(recoveryCtx, api, observed)
+ if err != nil {
+ return policyReplacementResult{err: err}
+ }
switch {
- case desired.matches(observed):
+ case observation.matches:
return policyReplacementResult{
desired: true,
observed: &observed,
warning: fmt.Sprintf("Dropping the previous %s returned an error, but reconciliation found the exact replacement already attached.", previous.label),
}
- case previous.matches(observed):
+ case previousMatches:
return policyReplacementResult{observed: &observed, err: fmt.Errorf("dropping the previous %s failed and reconciliation confirmed that it remains attached: %w", previous.label, dropErr)}
default:
return policyReplacementResult{observed: &observed, err: fmt.Errorf("dropping the previous %s returned an error (%s), and reconciliation found unexpected policy content for the same identity", previous.label, dropErr)}
@@ -621,21 +725,25 @@ func replacePolicyWithReconciliation(ctx context.Context, api *client.Client, pr
mutationCtx = context.WithoutCancel(ctx)
}
- created := createPolicyWithReconciliation(mutationCtx, api, desired)
+ created := createPolicyWithReconciliation(mutationCtx, api, desired, true)
if created.accepted {
return policyReplacementResult{desired: true, observed: created.observed, warning: created.warning, err: created.err}
}
if created.observed != nil {
- if previous.matches(*created.observed) {
+ matches, err := previous.matchesWithSchema(ctx, api, *created.observed)
+ if err != nil {
+ return policyReplacementResult{observed: created.observed, err: err}
+ }
+ if matches {
return policyReplacementResult{observed: created.observed, err: fmt.Errorf("creating the replacement failed, but reconciliation confirmed that the previous %s remains attached: %w", previous.label, created.err)}
}
return policyReplacementResult{observed: created.observed, err: fmt.Errorf("creating the replacement failed, and reconciliation found unexpected policy content for the same identity: %w", created.err)}
}
- recoveryCtx, cancel := mutationRecoveryContext(ctx)
+ recoveryCtx, cancel := mutationRecoveryContext(ctx, api)
defer cancel()
- restored := createPolicyWithReconciliation(recoveryCtx, api, previous)
+ restored := createPolicyWithReconciliation(recoveryCtx, api, previous, true)
if restored.accepted {
detail := fmt.Sprintf("creating the replacement failed (%s), so the previous %s was restored", created.err, previous.label)
if restored.err != nil {
@@ -644,7 +752,15 @@ func replacePolicyWithReconciliation(ctx context.Context, api *client.Client, pr
return policyReplacementResult{observed: restored.observed, err: fmt.Errorf("%s", detail)}
}
- if restored.observed != nil && desired.matches(*restored.observed) {
+ restoredDesired := false
+ if restored.observed != nil {
+ var err error
+ restoredDesired, err = desired.matchesWithSchema(recoveryCtx, api, *restored.observed)
+ if err != nil {
+ return policyReplacementResult{observed: restored.observed, err: err}
+ }
+ }
+ if restoredDesired {
return policyReplacementResult{
desired: true,
observed: restored.observed,
diff --git a/internal/provider/resource_table.go b/internal/provider/resource_table.go
index 05b2255..6a3129a 100644
--- a/internal/provider/resource_table.go
+++ b/internal/provider/resource_table.go
@@ -23,6 +23,7 @@ import (
"fmt"
"maps"
"net/url"
+ "slices"
"sort"
"strings"
@@ -45,24 +46,25 @@ type tableResource struct {
}
type tableResourceModel struct {
- ID types.String `tfsdk:"id"`
- CatalogID types.String `tfsdk:"catalog_id"`
- Database types.String `tfsdk:"database"`
- Name types.String `tfsdk:"name"`
- Fields types.List `tfsdk:"fields"`
- PartitionKeys types.List `tfsdk:"partition_keys"`
- PrimaryKeys types.List `tfsdk:"primary_keys"`
- Options types.Map `tfsdk:"options"`
- ServerOptions types.Map `tfsdk:"server_options"`
- Comment types.String `tfsdk:"comment"`
- SchemaID types.Int64 `tfsdk:"schema_id"`
- Path types.String `tfsdk:"path"`
- IsExternal types.Bool `tfsdk:"is_external"`
- Owner types.String `tfsdk:"owner"`
- CreatedAt types.Int64 `tfsdk:"created_at"`
- CreatedBy types.String `tfsdk:"created_by"`
- UpdatedAt types.Int64 `tfsdk:"updated_at"`
- UpdatedBy types.String `tfsdk:"updated_by"`
+ ID types.String `tfsdk:"id"`
+ ServerID types.String `tfsdk:"server_id"`
+ Database types.String `tfsdk:"database"`
+ Name types.String `tfsdk:"name"`
+ Fields types.List `tfsdk:"fields"`
+ PartitionKeys types.List `tfsdk:"partition_keys"`
+ PrimaryKeys types.List `tfsdk:"primary_keys"`
+ Options types.Map `tfsdk:"options"`
+ AllowReplacement types.Bool `tfsdk:"allow_replacement"`
+ ServerOptions types.Map `tfsdk:"server_options"`
+ Comment types.String `tfsdk:"comment"`
+ SchemaID types.Int64 `tfsdk:"schema_id"`
+ Path types.String `tfsdk:"path"`
+ IsExternal types.Bool `tfsdk:"is_external"`
+ Owner types.String `tfsdk:"owner"`
+ CreatedAt types.Int64 `tfsdk:"created_at"`
+ CreatedBy types.String `tfsdk:"created_by"`
+ UpdatedAt types.Int64 `tfsdk:"updated_at"`
+ UpdatedBy types.String `tfsdk:"updated_by"`
}
func NewTableResource() resource.Resource {
@@ -85,44 +87,99 @@ func (r *tableResource) Configure(_ context.Context, req resource.ConfigureReque
}
func (r *tableResource) ModifyPlan(ctx context.Context, req resource.ModifyPlanRequest, resp *resource.ModifyPlanResponse) {
- if req.State.Raw.IsNull() || req.Plan.Raw.IsNull() {
+ if req.Plan.Raw.IsNull() {
return
}
var config, state, plan tableResourceModel
resp.Diagnostics.Append(req.Config.Get(ctx, &config)...)
- resp.Diagnostics.Append(req.State.Get(ctx, &state)...)
+ if !req.State.Raw.IsNull() {
+ resp.Diagnostics.Append(req.State.Get(ctx, &state)...)
+ }
resp.Diagnostics.Append(req.Plan.Get(ctx, &plan)...)
if resp.Diagnostics.HasError() {
return
}
- var configuredFields, stateFields, plannedFields []tableFieldModel
- resp.Diagnostics.Append(config.Fields.ElementsAs(ctx, &configuredFields, false)...)
- resp.Diagnostics.Append(state.Fields.ElementsAs(ctx, &stateFields, false)...)
- resp.Diagnostics.Append(plan.Fields.ElementsAs(ctx, &plannedFields, false)...)
- if resp.Diagnostics.HasError() {
+ stabilizeTableKeys(ctx, config, state, &plan, &resp.Diagnostics)
+ replacementPaths := make([]path.Path, 0)
+ if tableFieldsInspectable(config.Fields) && tableFieldsInspectable(plan.Fields) {
+ var configuredFields, stateFields, plannedFields []tableFieldModel
+ resp.Diagnostics.Append(config.Fields.ElementsAs(ctx, &configuredFields, false)...)
+ if !req.State.Raw.IsNull() {
+ resp.Diagnostics.Append(state.Fields.ElementsAs(ctx, &stateFields, false)...)
+ }
+ resp.Diagnostics.Append(plan.Fields.ElementsAs(ctx, &plannedFields, false)...)
+ if resp.Diagnostics.HasError() {
+ return
+ }
+ if len(configuredFields) != len(plannedFields) {
+ resp.Diagnostics.AddError("Unable to stabilize Paimon field identities", "The configured and planned field lists have different lengths. Please report this issue to the provider developers.")
+
+ return
+ }
+
+ stabilizePlannedFieldIdentities(configuredFields, stateFields, plannedFields)
+ if !req.State.Raw.IsNull() {
+ validateAddedFieldIDs(stateFields, configuredFields, &resp.Diagnostics)
+ }
+ stabilizeFieldNullability(ctx, configuredFields, stateFields, plannedFields, plan, state, &resp.Diagnostics)
+ plan.Fields = fieldsValueFromModels(ctx, plannedFields, &resp.Diagnostics)
+ if resp.Diagnostics.HasError() {
+ return
+ }
+ if !req.State.Raw.IsNull() {
+ keyFields := knownTableKeyNames(state.PartitionKeys, state.PrimaryKeys, plan.PartitionKeys, plan.PrimaryKeys)
+ if compositeFieldTypesRequireReplace(stateFields, plannedFields) ||
+ keyFieldTypesRequireReplace(stateFields, plannedFields, keyFields) ||
+ newNonNullableFieldsRequireReplace(stateFields, plannedFields) {
+ replacementPaths = append(replacementPaths, path.Root("fields"))
+ }
+ }
+
+ }
+ resp.Diagnostics.Append(resp.Plan.Set(ctx, &plan)...)
+ if req.State.Raw.IsNull() || resp.Diagnostics.HasError() {
return
}
- if len(configuredFields) != len(plannedFields) {
- resp.Diagnostics.AddError("Unable to stabilize Paimon field identities", "The configured and planned field lists have different lengths. Please report this issue to the provider developers.")
+ if !plan.PrimaryKeys.IsUnknown() && !state.PrimaryKeys.Equal(plan.PrimaryKeys) {
+ replacementPaths = append(replacementPaths, path.Root("options").AtMapKey("primary-key"))
+ }
+ // Unknown identity and partition values may differ from the current table.
+ // Keep their replacement decision behind the same opt-in as known changes.
+ if !state.PartitionKeys.Equal(plan.PartitionKeys) {
+ replacementPaths = append(replacementPaths, path.Root("partition_keys"))
+ }
+ if !state.Name.Equal(plan.Name) {
+ replacementPaths = append(replacementPaths, path.Root("name"))
+ }
+ if !state.Database.Equal(plan.Database) {
+ replacementPaths = append(replacementPaths, path.Root("database"))
+ }
+ if len(replacementPaths) > 0 {
+ if !plan.AllowReplacement.ValueBool() {
+ resp.Diagnostics.AddError("Destructive table change is disabled", "This table identity or schema change requires dropping and recreating the table and can delete its data. Use a data migration, or explicitly set allow_replacement = true to permit replacement.")
- return
+ return
+ }
+ resp.RequiresReplace = append(resp.RequiresReplace, replacementPaths...)
}
+}
- stabilizePlannedFieldIdentities(configuredFields, stateFields, plannedFields)
- keyFields := append(stringListFromValue(ctx, state.PartitionKeys, &resp.Diagnostics), stringListFromValue(ctx, state.PrimaryKeys, &resp.Diagnostics)...)
- keyFields = append(keyFields, stringListFromValue(ctx, plan.PartitionKeys, &resp.Diagnostics)...)
- keyFields = append(keyFields, stringListFromValue(ctx, plan.PrimaryKeys, &resp.Diagnostics)...)
- plan.Fields = fieldsValueFromModels(ctx, plannedFields, &resp.Diagnostics)
- if resp.Diagnostics.HasError() {
- return
+func validateAddedFieldIDs(previous, configured []tableFieldModel, diags *diag.Diagnostics) {
+ known := make(map[int64]struct{}, len(previous))
+ for _, field := range previous {
+ if !field.ID.IsNull() && !field.ID.IsUnknown() {
+ known[field.ID.ValueInt64()] = struct{}{}
+ }
}
- resp.Diagnostics.Append(resp.Plan.Set(ctx, &plan)...)
- if compositeFieldTypesRequireReplace(stateFields, plannedFields) ||
- keyFieldTypesRequireReplace(stateFields, plannedFields, keyFields) ||
- newNonNullableFieldsRequireReplace(stateFields, plannedFields) {
- resp.RequiresReplace = append(resp.RequiresReplace, path.Root("fields"))
+ for index, field := range configured {
+ if field.ID.IsNull() || field.ID.IsUnknown() {
+ continue
+ }
+ if _, exists := known[field.ID.ValueInt64()]; !exists {
+ diags.AddAttributeError(path.Root("fields").AtListIndex(index).AtName("id"), "New field IDs are assigned by Paimon", "Omit id when adding a field to an existing table. Explicit IDs identify existing fields for updates and renames; the add-column API assigns new IDs on the server.")
+ }
}
}
@@ -166,7 +223,7 @@ func (r *tableResource) Create(ctx context.Context, req resource.CreateRequest,
if resp.Diagnostics.HasError() {
return
}
- if err := r.client.CreateTable(ctx, plan.Database.ValueString(), plan.Name.ValueString(), tableSchema); err != nil {
+ if err := r.client.CreateTable(ctx, plan.Database.ValueString(), plan.Name.ValueString(), tableCreateSchema(ctx, plan, tableSchema, &resp.Diagnostics)); err != nil {
if !client.IsMutationOutcomeUncertain(err) {
resp.Diagnostics.AddError("Unable to create Paimon table", err.Error())
@@ -237,10 +294,12 @@ func (r *tableResource) Update(ctx context.Context, req resource.UpdateRequest,
}
beforeSchema := schemaFromResourceModel(ctx, &state, &resp.Diagnostics)
- afterSchema := schemaFromResourceModel(ctx, &plan, &resp.Diagnostics)
+ effectivePlan := tablePlanWithServerDefaults(plan, state)
+ afterSchema := schemaFromResourceModel(ctx, &effectivePlan, &resp.Diagnostics)
var plannedFields []tableFieldModel
resp.Diagnostics.Append(plan.Fields.ElementsAs(ctx, &plannedFields, false)...)
if !resp.Diagnostics.HasError() {
+ validateAddedFieldIDs(fieldModelsFromRemote(beforeSchema.Fields), plannedFields, &resp.Diagnostics)
if err := assignTemporaryIDsToNewFields(beforeSchema.Fields, plannedFields, afterSchema.Fields); err != nil {
resp.Diagnostics.AddError("Unable to plan Paimon table field identities", err.Error())
}
@@ -251,7 +310,13 @@ func (r *tableResource) Update(ctx context.Context, req resource.UpdateRequest,
if resp.Diagnostics.HasError() {
return
}
- removals, updates := diffTableOptions(before, after)
+ if !slices.Equal(beforeSchema.PrimaryKeys, afterSchema.PrimaryKeys) || !slices.Equal(beforeSchema.PartitionKeys, afterSchema.PartitionKeys) {
+ resp.Diagnostics.AddError("Table key change requires replacement", "Changing table keys must be planned as a replacement; no in-place key mutation was sent.")
+
+ return
+ }
+ baseline := mapFromValue(ctx, effectiveManagedTableOptions(state.Options, plan.Options, tableOptionsWithPrimaryKeys(ctx, state.ServerOptions, state.PrimaryKeys, &resp.Diagnostics)), &resp.Diagnostics)
+ removals, updates := diffTableOptions(baseline, after)
sort.Strings(removals)
updateKeys := make([]string, 0, len(updates))
for key := range updates {
@@ -279,13 +344,19 @@ func (r *tableResource) Update(ctx context.Context, req resource.UpdateRequest,
if !state.Comment.Equal(plan.Comment) {
changes = append(changes, client.SchemaChange{"action": "updateComment", "comment": optionalStringPointer(plan.Comment)})
}
+ stableCtx := context.WithoutCancel(ctx)
+ resp.Diagnostics.Append(resp.State.Set(stableCtx, &state)...)
+ if resp.Diagnostics.HasError() {
+ return
+ }
+ ready := func(observed *client.Table) bool {
+ return tableMatchesPlannedSchema(observed, afterSchema, false, false) && optionsConverged(observed.Schema.Options, nil, removals)
+ }
if len(changes) > 0 {
if err := r.client.AlterTable(ctx, plan.Database.ValueString(), plan.Name.ValueString(), changes); err != nil {
if client.IsMutationOutcomeUncertain(err) {
- table, recoveryErr := r.lookupAfterMutation(ctx, plan.Database.ValueString(), plan.Name.ValueString(), func(observed *client.Table) bool {
- return tableMatchesPlannedSchema(observed, afterSchema, false, false)
- })
+ table, recoveryErr := r.lookupAfterMutation(ctx, plan.Database.ValueString(), plan.Name.ValueString(), ready)
if recoveryErr == nil {
stableCtx := context.WithoutCancel(ctx)
setTableResourceModel(stableCtx, &plan, table, &resp.Diagnostics)
@@ -300,17 +371,9 @@ func (r *tableResource) Update(ctx context.Context, req resource.UpdateRequest,
return
}
}
- stableCtx := context.WithoutCancel(ctx)
- plan.ID = types.StringValue(tableID(plan.Database.ValueString(), plan.Name.ValueString()))
- resp.Diagnostics.Append(resp.State.Set(stableCtx, &plan)...)
- if resp.Diagnostics.HasError() {
- return
- }
- table, err := r.lookupAfterMutation(ctx, plan.Database.ValueString(), plan.Name.ValueString(), func(observed *client.Table) bool {
- return tableMatchesPlannedSchema(observed, afterSchema, false, false)
- })
+ table, err := r.lookupAfterMutation(ctx, plan.Database.ValueString(), plan.Name.ValueString(), ready)
if err != nil {
- resp.Diagnostics.AddError("Unable to verify Paimon table after update", err.Error()+". Terraform retained the planned table state.")
+ resp.Diagnostics.AddError("Unable to verify Paimon table after update", err.Error()+". Terraform retained the previous state so unverified removals remain managed.")
return
}
@@ -342,7 +405,7 @@ func (r *tableResource) ImportState(ctx context.Context, req resource.ImportStat
}
func (r *tableResource) lookupAfterMutation(ctx context.Context, database, name string, ready func(*client.Table) bool) (*client.Table, error) {
- recoveryCtx, cancel := mutationRecoveryContext(ctx)
+ recoveryCtx, cancel := mutationRecoveryContext(ctx, r.client)
defer cancel()
table, found, converged, err := retryLookupUntil(recoveryCtx, func(attemptCtx context.Context) (*client.Table, bool, error) {
observed, lookupErr := r.client.GetTable(attemptCtx, database, name)
@@ -386,7 +449,16 @@ func tableMatchesPlannedSchema(table *client.Table, expected client.Schema, comp
}
}
for key, value := range expected.Options {
- if table.Schema.Options[key] != value {
+ observed, exists := table.Schema.Options[key]
+ if key == "type" {
+ if !exists {
+ observed = "table"
+ }
+ if strings.EqualFold(observed, value) {
+ continue
+ }
+ }
+ if !exists || observed != value {
return false
}
}
@@ -404,13 +476,16 @@ func setTableResourceModel(ctx context.Context, model *tableResourceModel, table
name = model.Name.ValueString()
}
model.ID = types.StringValue(tableID(database, name))
- model.CatalogID = types.StringValue(table.ID)
+ model.ServerID = types.StringValue(table.ID)
model.Database = types.StringValue(database)
model.Name = types.StringValue(name)
model.Fields = resourceFieldsValueFromRemote(ctx, model.Fields, table.Schema.Fields, diags)
model.PartitionKeys = stringListValue(ctx, table.Schema.PartitionKeys, diags)
model.PrimaryKeys = stringListValue(ctx, table.Schema.PrimaryKeys, diags)
- model.Options = syncManagedTableOptions(ctx, model.Options, table.Schema.Options, diags)
+ model.Options = syncManagedTableOptions(ctx, model.Options, tableOptionsMapWithPrimaryKeys(table.Schema.Options, table.Schema.PrimaryKeys), diags)
+ if model.AllowReplacement.IsNull() {
+ model.AllowReplacement = types.BoolValue(false)
+ }
model.ServerOptions = stringMapValue(ctx, table.Schema.Options, diags)
model.Comment = stringValueFromPointer(table.Schema.Comment)
model.SchemaID = types.Int64Value(table.SchemaID)
diff --git a/internal/provider/table_options.go b/internal/provider/table_options.go
index 55e617e..cf0ac3c 100644
--- a/internal/provider/table_options.go
+++ b/internal/provider/table_options.go
@@ -19,9 +19,11 @@ package provider
import (
"context"
+ "slices"
"strings"
"github.com/hashicorp/terraform-plugin-framework/diag"
+ "github.com/hashicorp/terraform-plugin-framework/path"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/mapplanmodifier"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier"
"github.com/hashicorp/terraform-plugin-framework/schema/validator"
@@ -57,11 +59,11 @@ var immutableTableOptions = map[string]struct{}{
type reservedTableOptionsValidator struct{}
func (reservedTableOptionsValidator) Description(context.Context) string {
- return "must configure primary keys and partitions with their first-class attributes"
+ return "must configure partitions with partition_keys"
}
func (reservedTableOptionsValidator) MarkdownDescription(context.Context) string {
- return "must configure primary keys and partitions with `primary_keys` and `partition_keys`"
+ return "must configure partitions with `partition_keys`"
}
func (reservedTableOptionsValidator) ValidateMap(_ context.Context, req validator.MapRequest, resp *validator.MapResponse) {
@@ -69,7 +71,7 @@ func (reservedTableOptionsValidator) ValidateMap(_ context.Context, req validato
return
}
reserved := make([]string, 0, 2)
- for _, key := range []string{"partition", "primary-key"} {
+ for _, key := range []string{"partition"} {
if _, exists := req.ConfigValue.Elements()[key]; exists {
reserved = append(reserved, key)
}
@@ -78,13 +80,49 @@ func (reservedTableOptionsValidator) ValidateMap(_ context.Context, req validato
resp.Diagnostics.AddAttributeError(
req.Path,
"Reserved Paimon table option",
- "Do not configure "+strings.Join(reserved, ", ")+" in options. Use partition_keys and primary_keys instead.",
+ "Do not configure "+strings.Join(reserved, ", ")+" in options. Use partition_keys instead; primary keys use options[\"primary-key\"].",
)
}
}
-func immutableTableOptionsRequiresReplace(_ context.Context, req planmodifier.MapRequest, resp *mapplanmodifier.RequiresReplaceIfFuncResponse) {
- resp.RequiresReplace = immutableTableOptionsChanged(req.StateValue, req.PlanValue)
+func immutableTableOptionsRequiresReplace(ctx context.Context, req planmodifier.MapRequest, resp *mapplanmodifier.RequiresReplaceIfFuncResponse) {
+ var remote types.Map
+ var allowed types.Bool
+ var keys types.List
+ resp.Diagnostics.Append(req.State.GetAttribute(ctx, path.Root("server_options"), &remote)...)
+ resp.Diagnostics.Append(req.Plan.GetAttribute(ctx, path.Root("allow_replacement"), &allowed)...)
+ if resp.Diagnostics.HasError() {
+ return
+ }
+ resp.Diagnostics.Append(req.State.GetAttribute(ctx, path.Root("primary_keys"), &keys)...)
+ before := effectiveManagedTableOptions(req.StateValue, req.PlanValue, tableOptionsWithPrimaryKeys(ctx, remote, keys, &resp.Diagnostics))
+ if immutableTableOptionsChanged(before, req.PlanValue) {
+ if !allowed.ValueBool() {
+ resp.Diagnostics.AddAttributeError(req.Path, "Destructive table change is disabled", "Changing an immutable option requires dropping and recreating the table and can delete its data. Use a data migration, or explicitly set allow_replacement = true to permit replacement.")
+
+ return
+ }
+ resp.RequiresReplace = true
+ }
+}
+
+// Starting to manage an existing option is not a remote change. Only include
+// newly managed keys so unrelated server options never become removal targets.
+func effectiveManagedTableOptions(before, after, remote types.Map) types.Map {
+ if before.IsUnknown() || after.IsUnknown() || remote.IsUnknown() {
+ return before
+ }
+ values := before.Elements()
+ remoteValues := remote.Elements()
+ for key := range after.Elements() {
+ if _, managed := values[key]; !managed {
+ if value, exists := remoteValues[key]; exists {
+ values[key] = value
+ }
+ }
+ }
+
+ return types.MapValueMust(types.StringType, values)
}
func immutableTableOptionsChanged(before, after types.Map) bool {
@@ -94,6 +132,13 @@ func immutableTableOptionsChanged(before, after types.Map) bool {
if !beforeKnown || !afterKnown {
continue
}
+ if key == "primary-key" {
+ if !slices.Equal(parsePrimaryKeyOption(beforeValue), parsePrimaryKeyOption(afterValue)) {
+ return true
+ }
+
+ continue
+ }
if key == "type" {
if !beforeExists {
beforeValue = "table"
@@ -135,7 +180,19 @@ func knownTableOption(options types.Map, key string) (string, bool, bool) {
}
func diffTableOptions(before, after map[string]string) ([]string, map[string]string) {
- removals, updates := diffOptions(before, after)
+ filteredBefore := make(map[string]string, len(before))
+ filteredAfter := make(map[string]string, len(after))
+ for key, value := range before {
+ if key != "primary-key" {
+ filteredBefore[key] = value
+ }
+ }
+ for key, value := range after {
+ if key != "primary-key" {
+ filteredAfter[key] = value
+ }
+ }
+ removals, updates := diffOptions(filteredBefore, filteredAfter)
beforeType, beforeExists := before["type"]
if !beforeExists {
beforeType = "table"
@@ -163,6 +220,7 @@ func syncManagedTableOptions(ctx context.Context, managed types.Map, remote map[
if managed.IsNull() || managed.IsUnknown() || diags.HasError() {
return synced
}
+ synced = retainPrimaryKeyOptionSpelling(managed, synced, remote)
managedOptions := mapFromValue(ctx, managed, diags)
configuredType, managesType := managedOptions["type"]
if !managesType || diags.HasError() {
diff --git a/internal/provider/table_plan.go b/internal/provider/table_plan.go
new file mode 100644
index 0000000..0638713
--- /dev/null
+++ b/internal/provider/table_plan.go
@@ -0,0 +1,206 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package provider
+
+import (
+ "context"
+ "slices"
+ "strconv"
+ "strings"
+
+ "github.com/apache/terraform-provider-paimon/internal/client"
+ "github.com/hashicorp/terraform-plugin-framework/diag"
+ "github.com/hashicorp/terraform-plugin-framework/types"
+)
+
+// Match Schema.normalizePrimaryKeys: trim tokens and discard empty entries.
+func parsePrimaryKeyOption(value string) []string {
+ keys := make([]string, 0)
+ for _, key := range strings.Split(value, ",") {
+ if key = strings.TrimSpace(key); key != "" {
+ keys = append(keys, key)
+ }
+ }
+
+ return keys
+}
+
+func tableOptionsMapWithPrimaryKeys(options map[string]string, keys []string) map[string]string {
+ logical := make(map[string]string, len(options)+1)
+ for key, value := range options {
+ logical[key] = value
+ }
+ logical["primary-key"] = strings.Join(keys, ",")
+
+ return logical
+}
+
+func tableOptionsWithPrimaryKeys(ctx context.Context, options types.Map, keys types.List, diags *diag.Diagnostics) types.Map {
+ if options.IsUnknown() || keys.IsUnknown() {
+ return types.MapUnknown(types.StringType)
+ }
+
+ return stringMapValue(ctx, tableOptionsMapWithPrimaryKeys(mapFromValue(ctx, options, diags), stringListFromValue(ctx, keys, diags)), diags)
+}
+
+func stabilizeTableKeys(ctx context.Context, config, state tableResourceModel, plan *tableResourceModel, diags *diag.Diagnostics) {
+ value, configured, known := knownTableOption(config.Options, "primary-key")
+ switch {
+ case !known:
+ plan.PrimaryKeys = types.ListUnknown(types.StringType)
+ case configured:
+ plan.PrimaryKeys = stringListValue(ctx, parsePrimaryKeyOption(value), diags)
+ default:
+ _, wasManaged, _ := knownTableOption(state.Options, "primary-key")
+ if wasManaged || state.PrimaryKeys.IsNull() {
+ plan.PrimaryKeys = stringListValue(ctx, nil, diags)
+ } else {
+ plan.PrimaryKeys = state.PrimaryKeys
+ }
+ }
+ if config.PartitionKeys.IsNull() {
+ if state.PartitionKeys.IsNull() {
+ plan.PartitionKeys = stringListValue(ctx, nil, diags)
+ } else {
+ plan.PartitionKeys = state.PartitionKeys
+ }
+ }
+}
+
+func tablePlanWithServerDefaults(plan, state tableResourceModel) tableResourceModel {
+ effective := plan
+ values := state.ServerOptions.Elements()
+ for _, key := range []string{"primary-key.nullable"} {
+ _, beforeManaged, _ := knownTableOption(state.Options, key)
+ _, afterManaged, known := knownTableOption(plan.Options, key)
+ if beforeManaged && !afterManaged && known {
+ delete(values, key)
+ }
+ }
+ effective.ServerOptions = types.MapValueMust(types.StringType, values)
+
+ return effective
+}
+
+func stabilizeFieldNullability(ctx context.Context, configured, previous, planned []tableFieldModel, plan, state tableResourceModel, diags *diag.Diagnostics) {
+ if plan.PrimaryKeys.IsUnknown() {
+ return
+ }
+ keys := stringListFromValue(ctx, plan.PrimaryKeys, diags)
+ effective := tablePlanWithServerDefaults(plan, state)
+ nullableValue, hasNullable, nullableKnown := knownTableOption(plan.Options, "primary-key.nullable")
+ if !hasNullable && nullableKnown {
+ nullableValue, _, nullableKnown = knownTableOption(effective.ServerOptions, "primary-key.nullable")
+ }
+ if !nullableKnown {
+ return
+ }
+ pkNullable := false
+ if nullableValue != "" {
+ var err error
+ pkNullable, err = strconv.ParseBool(nullableValue)
+ if err != nil {
+ diags.AddError("Invalid primary-key.nullable option", "Expected true or false.")
+
+ return
+ }
+ }
+ byID := make(map[int64]tableFieldModel, len(previous))
+ for _, field := range previous {
+ if !field.ID.IsNull() && !field.ID.IsUnknown() {
+ byID[field.ID.ValueInt64()] = field
+ }
+ }
+ for i := range planned {
+ if !configured[i].Nullable.IsNull() || configured[i].Type.IsUnknown() || configured[i].Name.IsUnknown() {
+ continue
+ }
+ switch {
+ case slices.Contains(keys, planned[i].Name.ValueString()):
+ planned[i].Nullable = types.BoolValue(pkNullable)
+ case !planned[i].ID.IsNull() && !planned[i].ID.IsUnknown():
+ if old, ok := byID[planned[i].ID.ValueInt64()]; ok {
+ planned[i].Nullable = old.Nullable
+
+ continue
+ }
+ _, nullable := splitFieldType(client.DataType(planned[i].Type.ValueString()))
+ planned[i].Nullable = types.BoolValue(nullable)
+ default:
+ _, nullable := splitFieldType(client.DataType(planned[i].Type.ValueString()))
+ planned[i].Nullable = types.BoolValue(nullable)
+ }
+ }
+}
+
+// Send the option through Java's normalization path without also declaring DDL keys.
+func tableCreateSchema(ctx context.Context, plan tableResourceModel, normalized client.Schema, diags *diag.Diagnostics) client.Schema {
+ options := mapFromValue(ctx, plan.Options, diags)
+ if value, exists := options["primary-key"]; exists {
+ normalized.Options = make(map[string]string, len(normalized.Options)+1)
+ for key, option := range options {
+ normalized.Options[key] = option
+ }
+ normalized.Options["primary-key"] = value
+ normalized.PrimaryKeys = []string{}
+ }
+
+ return normalized
+}
+
+func retainPrimaryKeyOptionSpelling(managed, synced types.Map, remote map[string]string) types.Map {
+ value, exists, known := knownTableOption(managed, "primary-key")
+ if !exists || !known || !slices.Equal(parsePrimaryKeyOption(value), parsePrimaryKeyOption(remote["primary-key"])) {
+ return synced
+ }
+ values := synced.Elements()
+ values["primary-key"] = types.StringValue(value)
+
+ return types.MapValueMust(types.StringType, values)
+}
+
+// An entire list or object may depend on another resource. Defer field planning
+// until its shape is known; primitive unknown attributes remain inspectable.
+func tableFieldsInspectable(fields types.List) bool {
+ if fields.IsUnknown() || fields.IsNull() {
+ return false
+ }
+ for _, field := range fields.Elements() {
+ if field.IsUnknown() || field.IsNull() {
+ return false
+ }
+ }
+
+ return true
+}
+
+// Replacement checks can inspect resolved key names while other list elements
+// remain unknown. Unknown names are checked when Terraform resolves the plan.
+func knownTableKeyNames(keyLists ...types.List) []string {
+ var names []string
+ for _, keys := range keyLists {
+ for _, element := range keys.Elements() {
+ key := element.(types.String)
+ if !key.IsNull() && !key.IsUnknown() {
+ names = append(names, key.ValueString())
+ }
+ }
+ }
+
+ return names
+}
diff --git a/internal/provider/table_schema.go b/internal/provider/table_schema.go
index b229d3b..abaa6f5 100644
--- a/internal/provider/table_schema.go
+++ b/internal/provider/table_schema.go
@@ -23,12 +23,13 @@ import (
"strings"
"github.com/apache/terraform-provider-paimon/internal/client"
+ "github.com/hashicorp/terraform-plugin-framework-validators/listvalidator"
"github.com/hashicorp/terraform-plugin-framework-validators/stringvalidator"
"github.com/hashicorp/terraform-plugin-framework/attr"
dschema "github.com/hashicorp/terraform-plugin-framework/datasource/schema"
"github.com/hashicorp/terraform-plugin-framework/diag"
rschema "github.com/hashicorp/terraform-plugin-framework/resource/schema"
- "github.com/hashicorp/terraform-plugin-framework/resource/schema/listplanmodifier"
+ "github.com/hashicorp/terraform-plugin-framework/resource/schema/booldefault"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/mapplanmodifier"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/planmodifier"
"github.com/hashicorp/terraform-plugin-framework/resource/schema/stringplanmodifier"
@@ -62,26 +63,26 @@ func tableFieldAttrTypes() map[string]attr.Type {
}
func tableResourceAttributes() map[string]rschema.Attribute {
+ // ModifyPlan handles identity and partition replacement together with the
+ // allow_replacement guard, including when configured values are unknown.
return map[string]rschema.Attribute{
"id": rschema.StringAttribute{
Description: "Stable URL-query identifier for the table identity.",
Computed: true,
PlanModifiers: []planmodifier.String{stringplanmodifier.UseStateForUnknown()},
},
- "catalog_id": rschema.StringAttribute{
+ "server_id": rschema.StringAttribute{
Description: "Server-assigned table identifier.",
Computed: true,
PlanModifiers: []planmodifier.String{stringplanmodifier.UseStateForUnknown()},
},
"database": rschema.StringAttribute{
- Description: "Database containing the table.",
- Required: true,
- PlanModifiers: []planmodifier.String{stringplanmodifier.RequiresReplace()},
+ Description: "Database containing the table.",
+ Required: true,
},
"name": rschema.StringAttribute{
- Description: "Table name.",
- Required: true,
- PlanModifiers: []planmodifier.String{stringplanmodifier.RequiresReplace()},
+ Description: "Table name.",
+ Required: true,
},
"fields": rschema.ListNestedAttribute{
Description: "Ordered table fields. Supported Paimon schema changes are applied in place using stable field IDs.",
@@ -89,18 +90,16 @@ func tableResourceAttributes() map[string]rschema.Attribute {
NestedObject: rschema.NestedAttributeObject{Attributes: tableFieldResourceAttributes()},
},
"partition_keys": rschema.ListAttribute{
- Description: "Ordered partition key field names.",
- Optional: true,
- Computed: true,
- ElementType: types.StringType,
- PlanModifiers: []planmodifier.List{listplanmodifier.RequiresReplaceIfConfigured()},
+ Description: "Ordered partition key field names.",
+ Optional: true,
+ Computed: true,
+ ElementType: types.StringType,
+ Validators: []validator.List{listvalidator.NoNullValues()},
},
"primary_keys": rschema.ListAttribute{
- Description: "Ordered primary key field names.",
- Optional: true,
- Computed: true,
- ElementType: types.StringType,
- PlanModifiers: []planmodifier.List{listplanmodifier.RequiresReplaceIfConfigured()},
+ Description: "Primary key field names returned by Paimon. Configure primary keys with options[\"primary-key\"], using comma-separated field names.",
+ Computed: true,
+ ElementType: types.StringType,
},
"options": rschema.MapAttribute{
Description: "Table options managed by Terraform. Options not declared here are preserved. Paimon options that are immutable after creation cause replacement when changed.",
@@ -109,6 +108,12 @@ func tableResourceAttributes() map[string]rschema.Attribute {
Validators: []validator.Map{reservedTableOptionsValidator{}},
PlanModifiers: []planmodifier.Map{mapplanmodifier.RequiresReplaceIf(immutableTableOptionsRequiresReplace, "replaces the table when an immutable Paimon option changes", "replaces the table when an immutable Paimon option changes")},
},
+ "allow_replacement": rschema.BoolAttribute{
+ Description: "Permit automatic table replacement for schema changes and immutable options that cannot be evolved safely. Replacement can delete all table data. Defaults to false; lifecycle.prevent_destroy provides additional protection.",
+ Optional: true,
+ Computed: true,
+ Default: booldefault.StaticBool(false),
+ },
"server_options": rschema.MapAttribute{
Description: "All table options returned by the REST Catalog.",
Computed: true,
@@ -212,7 +217,7 @@ func newNonNullableFieldsRequireReplace(before, after []tableFieldModel) bool {
func tableFieldResourceAttributes() map[string]rschema.Attribute {
return map[string]rschema.Attribute{
"id": rschema.Int64Attribute{
- Description: "Stable Paimon field ID between 0 and " + strconv.Itoa(maxPaimonFieldID) + ". The next available ID is assigned when omitted.",
+ Description: "Stable Paimon field ID between 0 and " + strconv.Itoa(maxPaimonFieldID) + ". May be specified on creation or to identify an existing field; omit when adding a field to an existing table because Paimon assigns its ID.",
Optional: true,
Computed: true,
},
@@ -223,12 +228,12 @@ func tableFieldResourceAttributes() map[string]rschema.Attribute {
Validators: []validator.String{stringvalidator.LengthAtLeast(1)},
},
"nullable": rschema.BoolAttribute{
- Description: "Whether the field accepts null values. Defaults to true when omitted.",
+ Description: "Whether the field accepts null values. Omitted values retain existing field nullability; new non-key fields default to true. Primary keys follow primary-key.nullable (false by default).",
Optional: true,
Computed: true,
},
"description": rschema.StringAttribute{Description: "Field description.", Optional: true},
- "default_value": rschema.StringAttribute{Description: "Field default expression.", Optional: true},
+ "default_value": rschema.StringAttribute{Description: "Constant default value as a string, converted to the field type by Paimon; SQL expressions are not evaluated. Null removes the default; an empty string is a distinct value.", Optional: true},
"nested_field_ids": rschema.MapAttribute{
Description: "Stable nested ROW field IDs keyed by escaped field path. Populated by the REST Catalog.",
Computed: true,
@@ -239,10 +244,10 @@ func tableFieldResourceAttributes() map[string]rschema.Attribute {
func tableDataSourceAttributes() map[string]dschema.Attribute {
return map[string]dschema.Attribute{
- "id": dschema.StringAttribute{Description: "Stable URL-query identifier for the table identity.", Computed: true},
- "catalog_id": dschema.StringAttribute{Description: "Server-assigned table identifier.", Computed: true},
- "database": dschema.StringAttribute{Description: "Database containing the table.", Required: true},
- "name": dschema.StringAttribute{Description: "Table name.", Required: true},
+ "id": dschema.StringAttribute{Description: "Stable URL-query identifier for the table identity.", Computed: true},
+ "server_id": dschema.StringAttribute{Description: "Server-assigned table identifier.", Computed: true},
+ "database": dschema.StringAttribute{Description: "Database containing the table.", Required: true},
+ "name": dschema.StringAttribute{Description: "Table name.", Required: true},
"fields": dschema.ListNestedAttribute{
Description: "Ordered table fields.",
Computed: true,
@@ -270,7 +275,7 @@ func tableFieldDataSourceAttributes() map[string]dschema.Attribute {
"type": dschema.StringAttribute{Description: "Canonical Paimon SQL data type.", Computed: true},
"nullable": dschema.BoolAttribute{Description: "Whether the field accepts null values.", Computed: true},
"description": dschema.StringAttribute{Description: "Field description.", Computed: true},
- "default_value": dschema.StringAttribute{Description: "Field default expression.", Computed: true},
+ "default_value": dschema.StringAttribute{Description: "Constant default represented as a string. Null means no default; an empty string is a distinct constant.", Computed: true},
"nested_field_ids": dschema.MapAttribute{
Description: "Stable nested ROW field IDs keyed by escaped field path.",
Computed: true,
@@ -286,14 +291,20 @@ func schemaFromResourceModel(ctx context.Context, model *tableResourceModel, dia
if diags.HasError() {
return client.Schema{}
}
- if _, exists := options["primary-key"]; exists {
- diags.AddError("Reserved table option", "Configure primary_keys instead of the primary-key table option.")
+ if configured, exists := options["primary-key"]; exists {
+ primaryKeys = parsePrimaryKeyOption(configured)
+ delete(options, "primary-key")
}
if _, exists := options["partition"]; exists {
diags.AddError("Reserved table option", "Configure partition_keys instead of the partition table option.")
}
primaryKeyNullable := false
- if configured, exists := options["primary-key.nullable"]; exists {
+ configured, exists := options["primary-key.nullable"]
+ if !exists && !model.ServerOptions.IsNull() && !model.ServerOptions.IsUnknown() {
+ serverOptions := mapFromValue(ctx, model.ServerOptions, diags)
+ configured, exists = serverOptions["primary-key.nullable"]
+ }
+ if exists {
parsed, err := strconv.ParseBool(configured)
if err != nil {
diags.AddError("Invalid primary-key.nullable option", "Expected a boolean value, got: "+configured)
@@ -440,10 +451,11 @@ func resourceFieldsValueFromRemote(ctx context.Context, managed types.List, fiel
if managedModels[index].Name.ValueString() != models[index].Name.ValueString() {
continue
}
- if client.EquivalentDataTypes(
- client.DataType(managedModels[index].Type.ValueString()),
- client.DataType(models[index].Type.ValueString()),
- ) {
+ managedType, managedNullable := splitFieldType(client.DataType(managedModels[index].Type.ValueString()))
+ if !managedNullable && models[index].Nullable.ValueBool() {
+ continue
+ }
+ if client.EquivalentDataTypes(managedType, client.DataType(models[index].Type.ValueString())) {
models[index].Type = managedModels[index].Type
}
}