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 555bff5 Fix REST security and table schema contracts (#8)
555bff5 is described below
commit 555bff55d197db9a4f9fa2c13812447058dc920c
Author: Jingsong Lee <[email protected]>
AuthorDate: Mon Aug 24 15:27:37 2026 +0800
Fix REST security and table schema contracts (#8)
---
README.md | 7 +-
docs/index.md | 28 +-
docs/resources/table.md | 13 +-
examples/dlf-ecs/main.tf | 3 +-
examples/dlf-sts/main.tf | 3 +-
examples/dlf-token-file/main.tf | 3 +-
internal/client/client.go | 34 +-
internal/client/client_test.go | 200 ++++++++++-
internal/client/data_type.go | 651 ++++++++++++++++++++++++++++++++++++
internal/client/dlf_auth.go | 37 +-
internal/client/dlf_auth_test.go | 9 +
internal/client/models.go | 14 +-
internal/provider/provider_test.go | 333 ++++++++++++++++++
internal/provider/resource_table.go | 6 +-
internal/provider/table_options.go | 184 ++++++++++
internal/provider/table_schema.go | 112 ++++++-
16 files changed, 1580 insertions(+), 57 deletions(-)
diff --git a/README.md b/README.md
index 5bbbbcd..7951224 100644
--- a/README.md
+++ b/README.md
@@ -115,11 +115,8 @@ terraform import paimon_table.events analytics.events
Go 1.25 or newer is required.
```bash
-make fmt
-make test
-make build
+make check
```
See [`docs/index.md`](docs/index.md) for the full provider configuration and
-resource notes. The source comparison and implementation tradeoffs are recorded
-in [`docs/design.md`](docs/design.md).
+resource notes.
diff --git a/docs/index.md b/docs/index.md
index 840878c..bd83a92 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -42,10 +42,36 @@ provider "paimon" {
- `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
+token file, or an ECS RAM role. DLF Catalog requests also require `warehouse`.
+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.
The provider first calls `/v1/config`, merges server defaults, client values,
and server overrides in that order, and then uses the resulting `prefix` for
-catalog operations.
+catalog operations. Redirects are rejected so authentication headers and DLF
+signatures are never reused for a different URL.
## Resources and data sources
diff --git a/docs/resources/table.md b/docs/resources/table.md
index 84c32f1..9d0cea3 100644
--- a/docs/resources/table.md
+++ b/docs/resources/table.md
@@ -46,13 +46,20 @@ resource "paimon_table" "example" {
```
Each field supports `id`, `name`, `type`, `nullable`, `description`, and
-`default_value`. Field IDs are assigned by list position when omitted. Use
+`default_value`. Field IDs must be unique integers from 0 through 1073741822;
+the next available ID is assigned when omitted. Use
canonical Paimon SQL type strings such as `INT`, `BIGINT`, `STRING`,
`DECIMAL(12, 2)`, `ARRAY<STRING>`, or `ROW<item STRING>`.
`database`, `name`, `fields`, `partition_keys`, and `primary_keys` are
-replacement attributes. `options` and `comment` update in place. Unmanaged
-server options are preserved and exposed through `server_options`.
+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.
+
+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`.
Dropping a managed table can delete its data. Use `prevent_destroy` where
appropriate:
diff --git a/examples/dlf-ecs/main.tf b/examples/dlf-ecs/main.tf
index bba0a06..44fd1bb 100644
--- a/examples/dlf-ecs/main.tf
+++ b/examples/dlf-ecs/main.tf
@@ -22,7 +22,8 @@ terraform {
}
provider "paimon" {
- uri = "https://dlf.cn-hangzhou.aliyuncs.com";
+ uri = "https://dlfnext.cn-hangzhou.aliyuncs.com";
+ warehouse = "my_catalog"
token_provider = "dlf"
dlf_token_loader = "ecs"
diff --git a/examples/dlf-sts/main.tf b/examples/dlf-sts/main.tf
index b56bc95..9e65fc7 100644
--- a/examples/dlf-sts/main.tf
+++ b/examples/dlf-sts/main.tf
@@ -38,7 +38,8 @@ variable "dlf_security_token" {
}
provider "paimon" {
- uri = "https://dlf.cn-hangzhou.aliyuncs.com";
+ uri = "https://dlfnext.cn-hangzhou.aliyuncs.com";
+ warehouse = "my_catalog"
token_provider = "dlf"
dlf_access_key_id = var.dlf_access_key_id
diff --git a/examples/dlf-token-file/main.tf b/examples/dlf-token-file/main.tf
index 611323e..132cd77 100644
--- a/examples/dlf-token-file/main.tf
+++ b/examples/dlf-token-file/main.tf
@@ -22,7 +22,8 @@ terraform {
}
provider "paimon" {
- uri = "https://dlf.cn-hangzhou.aliyuncs.com";
+ uri = "https://dlfnext.cn-hangzhou.aliyuncs.com";
+ warehouse = "my_catalog"
token_provider = "dlf"
dlf_token_loader = "local_file"
dlf_token_path = "/run/secrets/dlf-sts.json"
diff --git a/internal/client/client.go b/internal/client/client.go
index 24310bf..bc64165 100644
--- a/internal/client/client.go
+++ b/internal/client/client.go
@@ -32,7 +32,10 @@ import (
"time"
)
-const userAgent = "terraform-provider-paimon"
+const (
+ userAgent = "terraform-provider-paimon"
+ maxAPIResponseBodySize = 16 << 20
+)
type Config struct {
URI string
@@ -96,10 +99,7 @@ func New(config Config) (*Client, error) {
return nil, errors.New("Paimon REST URI must not include a
query or fragment")
}
- httpClient := config.HTTPClient
- if httpClient == nil {
- httpClient = &http.Client{Timeout: 30 * time.Second}
- }
+ httpClient := noRedirectHTTPClient(config.HTTPClient)
configuredAuth :=
strings.ToLower(strings.TrimSpace(config.AuthProvider))
if configuredAuth == "" && config.Token != "" {
@@ -311,18 +311,36 @@ func (c *Client) doRaw(ctx context.Context, method
string, segments []string, qu
return apiErr
}
- if result == nil || response.StatusCode == http.StatusNoContent {
- _, _ = io.Copy(io.Discard, response.Body)
+ contents, err := io.ReadAll(io.LimitReader(response.Body,
maxAPIResponseBodySize+1))
+ if err != nil {
+ return errors.New("read Paimon REST response")
+ }
+ if len(contents) > maxAPIResponseBodySize {
+ return errors.New("Paimon REST response exceeded 16 MiB size
limit")
+ }
+ if result == nil || response.StatusCode == http.StatusNoContent {
return nil
}
- if err := json.NewDecoder(response.Body).Decode(result); err != nil {
+ if err := json.Unmarshal(contents, result); err != nil {
return fmt.Errorf("decode Paimon REST response: %w", err)
}
return nil
}
+func noRedirectHTTPClient(input *http.Client) *http.Client {
+ if input == nil {
+ input = &http.Client{Timeout: 30 * time.Second}
+ }
+ output := *input
+ output.CheckRedirect = func(_ *http.Request, _ []*http.Request) error {
+ return errors.New("Paimon REST API redirects are not allowed")
+ }
+
+ return &output
+}
+
func (c *Client) endpoint(segments []string, query url.Values) string {
endpoint := *c.baseURL
escapedPath := strings.TrimRight(endpoint.EscapedPath(), "/")
diff --git a/internal/client/client_test.go b/internal/client/client_test.go
index 2d684ea..dac103d 100644
--- a/internal/client/client_test.go
+++ b/internal/client/client_test.go
@@ -22,6 +22,7 @@ import (
"encoding/json"
"net/http"
"net/http/httptest"
+ "strings"
"sync/atomic"
"testing"
@@ -140,7 +141,144 @@ func TestDataTypeStructuredJSON(t *testing.T) {
encoded, err := json.Marshal(dataType)
require.NoError(t, err)
- assert.JSONEq(t, `"ROW<item ARRAY<STRING NOT NULL>> NOT NULL"`,
string(encoded))
+ assert.JSONEq(t, `{
+ "type":"ROW NOT NULL",
+
"fields":[{"id":0,"name":"item","type":{"type":"ARRAY","element":"STRING NOT
NULL"}}]
+ }`, string(encoded))
+}
+
+func TestSchemaMarshalAssignsUniqueNestedFieldIDs(t *testing.T) {
+ schema := Schema{
+ Fields: []Field{
+ {ID: 2, Name: "id", Type: DataType("BIGINT NOT NULL")},
+ {
+ 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)>"),
+ },
+ },
+ }
+
+ encoded, err := json.Marshal(schema)
+ require.NoError(t, err)
+ assert.JSONEq(t, `{
+ "fields":[
+ {"id":2,"name":"id","type":"BIGINT NOT NULL"},
+ {"id":7,"name":"payload","type":{
+ "type":"ROW",
+ "fields":[{
+ "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)"
+ }]
+ }}
+ ],
+ "partitionKeys":[],
+ "primaryKeys":[],
+ "options":{}
+ }`, string(encoded))
+}
+
+func TestDataTypeMarshalRejectsMalformedComposite(t *testing.T) {
+ _, err := json.Marshal(DataType("MAP<STRING>"))
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "expected key and value types")
+}
+
+func TestDataTypeMarshalSupportsEmptyRow(t *testing.T) {
+ encoded, err := json.Marshal(DataType("ROW<> NOT NULL"))
+ require.NoError(t, err)
+ assert.JSONEq(t, `{"type":"ROW NOT NULL","fields":[]}`, string(encoded))
+}
+
+func TestDataTypeMarshalKeepsComparisonsInNestedDefaults(t *testing.T) {
+ for _, test := range []struct {
+ name string
+ dataType DataType
+ greaterDefault string
+ lesserDefault string
+ }{
+ {
+ name: "parenthesized",
+ dataType: DataType("ROW<greater BOOLEAN DEFAULT
(1 > 0), lesser BOOLEAN DEFAULT (1 < 2)>"),
+ greaterDefault: "(1 > 0)",
+ lesserDefault: "(1 < 2)",
+ },
+ {
+ name: "unparenthesized",
+ dataType: DataType("ROW<greater BOOLEAN DEFAULT 1
> 0, lesser BOOLEAN DEFAULT 1 < 2>"),
+ greaterDefault: "1 > 0",
+ lesserDefault: "1 < 2",
+ },
+ {
+ name: "opaque composite keyword",
+ dataType: DataType("ROW<greater BOOLEAN DEFAULT
MAP > 0, lesser BOOLEAN DEFAULT MAP < 2>"),
+ greaterDefault: "MAP > 0",
+ lesserDefault: "MAP < 2",
+ },
+ } {
+ t.Run(test.name, func(t *testing.T) {
+ encoded, err := json.Marshal(test.dataType)
+ require.NoError(t, err)
+ var structured struct {
+ Fields []struct {
+ DefaultValue string
`json:"defaultValue"`
+ } `json:"fields"`
+ }
+ require.NoError(t, json.Unmarshal(encoded, &structured))
+ require.Len(t, structured.Fields, 2)
+ assert.Equal(t, test.greaterDefault,
structured.Fields[0].DefaultValue)
+ assert.Equal(t, test.lesserDefault,
structured.Fields[1].DefaultValue)
+ })
+ }
+}
+
+func TestEquivalentDataTypesNormalizesCompositeSpelling(t *testing.T) {
+ assert.True(t, EquivalentDataTypes(DataType("MAP<STRING,STRING>"),
DataType("MAP<STRING, STRING>")))
+ assert.True(t, EquivalentDataTypes(DataType("ROW<`item` STRING>"),
DataType("ROW<item STRING>")))
+ assert.False(t, EquivalentDataTypes(DataType("MAP<STRING, STRING>"),
DataType("MAP<STRING, BIGINT>")))
+}
+
+func TestDataTypeMarshalKeepsComparisonInNestedRowDefault(t *testing.T) {
+ encoded, err := json.Marshal(DataType("ROW<nested ROW<flag BOOLEAN
DEFAULT 1 > 0>, tail BOOLEAN DEFAULT 1 < 2>"))
+ require.NoError(t, err)
+ assert.JSONEq(t, `{
+ "type":"ROW",
+ "fields":[
+ {"id":0,"name":"nested","type":{"type":"ROW","fields":[
+
{"id":1,"name":"flag","type":"BOOLEAN","defaultValue":"1 > 0"}
+ ]}},
+
{"id":2,"name":"tail","type":"BOOLEAN","defaultValue":"1 < 2"}
+ ]
+ }`, string(encoded))
+}
+
+func TestSchemaMarshalRejectsInvalidFieldIDs(t *testing.T) {
+ _, err := json.Marshal(Schema{Fields: []Field{
+ {ID: maxPaimonFieldID, Name: "max", Type: DataType("STRING")},
+ }})
+ require.NoError(t, err)
+
+ _, err = json.Marshal(Schema{Fields: []Field{
+ {ID: 1, Name: "first", Type: DataType("STRING")},
+ {ID: 1, Name: "duplicate", Type: DataType("STRING")},
+ }})
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "field ID 1 is duplicated")
+
+ _, err = json.Marshal(Schema{Fields: []Field{
+ {ID: maxPaimonFieldID + 1, Name: "reserved", Type:
DataType("STRING")},
+ }})
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "must be between 0 and")
+
+ _, err = json.Marshal(Schema{Fields: []Field{
+ {ID: maxPaimonFieldID, Name: "row", Type: DataType("ROW<nested
STRING>")},
+ }})
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "nested field IDs exceed")
}
func TestNewRejectsInvalidURI(t *testing.T) {
@@ -148,6 +286,66 @@ func TestNewRejectsInvalidURI(t *testing.T) {
require.EqualError(t, err, "Paimon REST URI must use http or https")
}
+func TestClientRejectsRedirectWithoutForwardingCredentials(t *testing.T) {
+ var targetCalls atomic.Int32
+ target := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ targetCalls.Add(1)
+ assert.Empty(t, r.Header.Get("Authorization"))
+ w.WriteHeader(http.StatusOK)
+ }))
+ defer target.Close()
+
+ redirect := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ http.Redirect(w, r, target.URL+r.URL.Path,
http.StatusTemporaryRedirect)
+ }))
+ defer redirect.Close()
+
+ for name, config := range map[string]Config{
+ "bearer": {URI: redirect.URL, Token: "must-not-leak"},
+ "dlf": {
+ URI: redirect.URL,
+ AuthProvider: AuthProviderDLF,
+ DLF: &DLFConfig{
+ Region: "cn-hangzhou",
+ AccessKeyID: "must-not-leak-id",
+ AccessKeySecret: "must-not-leak-secret",
+ SecurityToken: "must-not-leak-sts",
+ SigningAlgorithm: DLFSigningDefault,
+ },
+ },
+ } {
+ t.Run(name, func(t *testing.T) {
+ api, err := New(config)
+ require.NoError(t, err)
+ _, err = api.GetDatabase(context.Background(),
"analytics")
+ require.Error(t, err)
+ assert.Contains(t, err.Error(), "redirects are not
allowed")
+ assert.NotContains(t, err.Error(), "must-not-leak")
+ assert.Equal(t, int32(0), targetCalls.Load())
+ })
+ }
+}
+
+func TestClientRejectsOversizedSuccessfulResponse(t *testing.T) {
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/v1/config" {
+ writeJSON(t, w, ConfigResponse{Defaults:
map[string]string{"prefix": "catalog"}})
+
+ return
+ }
+ w.Header().Set("Content-Type", "application/json")
+ _, _ = w.Write([]byte(`{"name":"`))
+ _, _ = w.Write([]byte(strings.Repeat("x",
maxAPIResponseBodySize)))
+ _, _ = w.Write([]byte(`"}`))
+ }))
+ defer server.Close()
+
+ api, err := New(Config{URI: server.URL})
+ require.NoError(t, err)
+ _, err = api.GetDatabase(context.Background(), "analytics")
+ require.EqualError(t, err, "Paimon REST response exceeded 16 MiB size
limit")
+}
+
func writeJSON(t *testing.T, w http.ResponseWriter, value any) {
t.Helper()
w.Header().Set("Content-Type", "application/json")
diff --git a/internal/client/data_type.go b/internal/client/data_type.go
new file mode 100644
index 0000000..897c9c6
--- /dev/null
+++ b/internal/client/data_type.go
@@ -0,0 +1,651 @@
+// 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 client
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "strconv"
+ "strings"
+ "unicode"
+)
+
+// Paimon reserves IDs at and above SpecialFields.SYSTEM_FIELD_ID_START.
+const maxPaimonFieldID = (1 << 30) - 2
+
+type structuredDataType struct {
+ Type string `json:"type"`
+ Element any `json:"element,omitempty"`
+ Key any `json:"key,omitempty"`
+ Value any `json:"value,omitempty"`
+ Fields *[]structuredField `json:"fields,omitempty"`
+ Length int `json:"length,omitempty"`
+}
+
+type structuredField struct {
+ ID int `json:"id"`
+ Name string `json:"name"`
+ Type any `json:"type"`
+ Description *string `json:"description,omitempty"`
+ DefaultValue *string `json:"defaultValue,omitempty"`
+}
+
+// MarshalJSON coordinates nested ROW field IDs across the complete schema.
+// Paimon requires nested fields to carry IDs when their parent field has one.
+func (s Schema) MarshalJSON() ([]byte, error) {
+ type schemaJSON struct {
+ Fields []structuredField `json:"fields"`
+ PartitionKeys []string `json:"partitionKeys"`
+ PrimaryKeys []string `json:"primaryKeys"`
+ Options map[string]string `json:"options"`
+ Comment *string `json:"comment,omitempty"`
+ }
+
+ nextFieldID := -1
+ usedFieldIDs := make(map[int]struct{}, len(s.Fields))
+ for _, field := range s.Fields {
+ if field.ID < 0 || field.ID > maxPaimonFieldID {
+ return nil, fmt.Errorf("Paimon field %q ID must be
between 0 and %d", field.Name, maxPaimonFieldID)
+ }
+ if _, duplicate := usedFieldIDs[field.ID]; duplicate {
+ return nil, fmt.Errorf("Paimon field ID %d is
duplicated", field.ID)
+ }
+ usedFieldIDs[field.ID] = struct{}{}
+ if field.ID > nextFieldID {
+ nextFieldID = field.ID
+ }
+ }
+ fields := make([]structuredField, 0, len(s.Fields))
+ for _, field := range s.Fields {
+ encodedType, err := encodeDataType(string(field.Type),
&nextFieldID)
+ if err != nil {
+ return nil, fmt.Errorf("encode Paimon field %q type:
%w", field.Name, err)
+ }
+ fields = append(fields, structuredField{
+ ID: field.ID,
+ Name: field.Name,
+ Type: encodedType,
+ Description: field.Description,
+ DefaultValue: field.DefaultValue,
+ })
+ }
+
+ return json.Marshal(schemaJSON{
+ Fields: fields,
+ PartitionKeys: nonNilSlice(s.PartitionKeys),
+ PrimaryKeys: nonNilSlice(s.PrimaryKeys),
+ Options: nonNilMap(s.Options),
+ Comment: s.Comment,
+ })
+}
+
+func encodeDataType(input string, nextFieldID *int) (any, error) {
+ typeName, notNull := stripNotNull(input)
+ root, body, composite, err := compositeTypeParts(typeName)
+ if err != nil {
+ return nil, err
+ }
+ if !composite {
+ return strings.TrimSpace(input), nil
+ }
+
+ serializedRoot := root
+ if notNull {
+ serializedRoot += " NOT NULL"
+ }
+ parts, err := splitTopLevel(body, ',')
+ if err != nil {
+ return nil, fmt.Errorf("invalid %s type %q: %w", root, input,
err)
+ }
+
+ switch root {
+ case "ARRAY", "MULTISET":
+ if len(parts) != 1 || strings.TrimSpace(parts[0]) == "" {
+ return nil, fmt.Errorf("invalid %s type %q: expected
one element type", root, input)
+ }
+ element, err := encodeDataType(parts[0], nextFieldID)
+ if err != nil {
+ return nil, err
+ }
+
+ return structuredDataType{Type: serializedRoot, Element:
element}, nil
+ case "MAP":
+ if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" ||
strings.TrimSpace(parts[1]) == "" {
+ return nil, fmt.Errorf("invalid MAP type %q: expected
key and value types", input)
+ }
+ key, err := encodeDataType(parts[0], nextFieldID)
+ if err != nil {
+ return nil, err
+ }
+ value, err := encodeDataType(parts[1], nextFieldID)
+ if err != nil {
+ return nil, err
+ }
+
+ return structuredDataType{Type: serializedRoot, Key: key,
Value: value}, nil
+ case "VECTOR":
+ if len(parts) != 2 || strings.TrimSpace(parts[0]) == "" {
+ return nil, fmt.Errorf("invalid VECTOR type %q:
expected element type and length", input)
+ }
+ element, err := encodeDataType(parts[0], nextFieldID)
+ if err != nil {
+ return nil, err
+ }
+ length, err := strconv.Atoi(strings.TrimSpace(parts[1]))
+ if err != nil || length <= 0 {
+ return nil, fmt.Errorf("invalid VECTOR type %q: length
must be a positive integer", input)
+ }
+
+ return structuredDataType{Type: serializedRoot, Element:
element, Length: length}, nil
+ case "ROW":
+ fields := make([]structuredField, 0, len(parts))
+ if len(parts) == 1 && parts[0] == "" {
+ return structuredDataType{Type: serializedRoot, Fields:
&fields}, nil
+ }
+ for _, part := range parts {
+ name, fieldType, description, defaultValue, err :=
parseRowField(part)
+ if err != nil {
+ return nil, fmt.Errorf("invalid ROW type %q:
%w", input, err)
+ }
+ if *nextFieldID >= maxPaimonFieldID {
+ return nil, fmt.Errorf("Paimon nested field IDs
exceed %d", maxPaimonFieldID)
+ }
+ *nextFieldID = *nextFieldID + 1
+ fieldID := *nextFieldID
+ encodedType, err := encodeDataType(fieldType,
nextFieldID)
+ if err != nil {
+ return nil, err
+ }
+ fields = append(fields, structuredField{
+ ID: fieldID,
+ Name: name,
+ Type: encodedType,
+ Description: description,
+ DefaultValue: defaultValue,
+ })
+ }
+
+ return structuredDataType{Type: serializedRoot, Fields:
&fields}, nil
+ default:
+ return nil, fmt.Errorf("unsupported composite Paimon type %q",
root)
+ }
+}
+
+// EquivalentDataTypes reports whether two SQL spellings encode to the same
+// language-neutral Paimon REST data type.
+func EquivalentDataTypes(left, right DataType) bool {
+ canonicalLeft, err := canonicalDataType(left)
+ if err != nil {
+ return false
+ }
+ canonicalRight, err := canonicalDataType(right)
+
+ return err == nil && canonicalLeft == canonicalRight
+}
+
+func canonicalDataType(value DataType) (DataType, error) {
+ encoded, err := json.Marshal(value)
+ if err != nil {
+ return "", err
+ }
+ var canonical DataType
+ if err := json.Unmarshal(encoded, &canonical); err != nil {
+ return "", err
+ }
+
+ return canonical, nil
+}
+
+func stripNotNull(input string) (string, bool) {
+ trimmed := strings.TrimSpace(input)
+ const suffix = " NOT NULL"
+ if len(trimmed) >= len(suffix) &&
strings.EqualFold(trimmed[len(trimmed)-len(suffix):], suffix) {
+ return strings.TrimSpace(trimmed[:len(trimmed)-len(suffix)]),
true
+ }
+
+ return trimmed, false
+}
+
+func compositeTypeParts(input string) (string, string, bool, error) {
+ trimmed := strings.TrimSpace(input)
+ opening := strings.IndexByte(trimmed, '<')
+ leading := trimmed
+ if opening >= 0 {
+ leading = strings.TrimSpace(trimmed[:opening])
+ }
+ root := strings.ToUpper(leading)
+ isCompositeRoot := root == "ARRAY" || root == "MAP" || root ==
"MULTISET" || root == "ROW" || root == "VECTOR"
+ if !isCompositeRoot {
+ return "", "", false, nil
+ }
+ if opening < 0 {
+ return "", "", false, fmt.Errorf("invalid %s type %q: missing
angle brackets", root, input)
+ }
+
+ closing, err := matchingAngleBracket(trimmed, opening)
+ if err != nil {
+ return "", "", false, fmt.Errorf("invalid %s type %q: %w",
root, input, err)
+ }
+ if strings.TrimSpace(trimmed[closing+1:]) != "" {
+ return "", "", false, fmt.Errorf("invalid %s type %q:
unexpected trailing content", root, input)
+ }
+
+ return root, trimmed[opening+1 : closing], true, nil
+}
+
+func matchingAngleBracket(input string, opening int) (int, error) {
+ depth := 0
+ parenDepth, squareDepth, braceDepth := 0, 0, 0
+ defaultAtDepth := make(map[int]bool)
+ inString := false
+ inIdentifier := false
+ for index := opening; index < len(input); index++ {
+ if !inString && !inIdentifier && parenDepth == 0 && squareDepth
== 0 && braceDepth == 0 && hasKeywordAt(input, index, "DEFAULT") {
+ defaultAtDepth[depth] = true
+ }
+ switch input[index] {
+ case '\'':
+ if inIdentifier {
+ continue
+ }
+ if inString && index+1 < len(input) && input[index+1]
== '\'' {
+ index++
+
+ continue
+ }
+ inString = !inString
+ case '`':
+ if inString {
+ continue
+ }
+ if inIdentifier && index+1 < len(input) &&
input[index+1] == '`' {
+ index++
+
+ continue
+ }
+ inIdentifier = !inIdentifier
+ case '<':
+ if !inString && !inIdentifier && !defaultAtDepth[depth]
&& parenDepth == 0 && squareDepth == 0 && braceDepth == 0 &&
isCompositeAngleOpening(input, index) {
+ depth++
+ }
+ case '>':
+ if !inString && !inIdentifier && parenDepth == 0 &&
squareDepth == 0 && braceDepth == 0 && isStructuralAngleClosing(input, index) {
+ delete(defaultAtDepth, depth)
+ depth--
+ if depth == 0 {
+ return index, nil
+ }
+ if depth < 0 {
+ return -1, errors.New("unexpected
closing angle bracket")
+ }
+ }
+ case '(':
+ if !inString && !inIdentifier {
+ parenDepth++
+ }
+ case ')':
+ if !inString && !inIdentifier {
+ parenDepth--
+ }
+ case '[':
+ if !inString && !inIdentifier {
+ squareDepth++
+ }
+ case ']':
+ if !inString && !inIdentifier {
+ squareDepth--
+ }
+ case '{':
+ if !inString && !inIdentifier {
+ braceDepth++
+ }
+ case '}':
+ if !inString && !inIdentifier {
+ braceDepth--
+ }
+ case ',':
+ if !inString && !inIdentifier && parenDepth == 0 &&
squareDepth == 0 && braceDepth == 0 {
+ delete(defaultAtDepth, depth)
+ }
+ }
+ if parenDepth < 0 || squareDepth < 0 || braceDepth < 0 {
+ return -1, errors.New("unbalanced delimiters")
+ }
+ }
+ if inString || inIdentifier || parenDepth != 0 || squareDepth != 0 ||
braceDepth != 0 {
+ return -1, errors.New("unbalanced delimiters or quotes")
+ }
+
+ return -1, errors.New("missing closing angle bracket")
+}
+
+func splitTopLevel(input string, separator byte) ([]string, error) {
+ parts := make([]string, 0, 2)
+ start := 0
+ angleDepth, parenDepth, squareDepth, braceDepth := 0, 0, 0, 0
+ defaultAtDepth := make(map[int]bool)
+ inString := false
+ inIdentifier := false
+ for index := 0; index < len(input); index++ {
+ character := input[index]
+ if !inString && !inIdentifier && parenDepth == 0 && squareDepth
== 0 && braceDepth == 0 && hasKeywordAt(input, index, "DEFAULT") {
+ defaultAtDepth[angleDepth] = true
+ }
+ switch character {
+ case '\'':
+ if inIdentifier {
+ continue
+ }
+ if inString && index+1 < len(input) && input[index+1]
== '\'' {
+ index++
+
+ continue
+ }
+ inString = !inString
+ case '`':
+ if inString {
+ continue
+ }
+ if inIdentifier && index+1 < len(input) &&
input[index+1] == '`' {
+ index++
+
+ continue
+ }
+ inIdentifier = !inIdentifier
+ case '<':
+ if !inString && !inIdentifier &&
!defaultAtDepth[angleDepth] && parenDepth == 0 && squareDepth == 0 &&
braceDepth == 0 && isCompositeAngleOpening(input, index) {
+ angleDepth++
+ }
+ case '>':
+ if !inString && !inIdentifier && parenDepth == 0 &&
squareDepth == 0 && braceDepth == 0 && isStructuralAngleClosing(input, index) {
+ delete(defaultAtDepth, angleDepth)
+ angleDepth--
+ }
+ case '(':
+ if !inString && !inIdentifier {
+ parenDepth++
+ }
+ case ')':
+ if !inString && !inIdentifier {
+ parenDepth--
+ }
+ case '[':
+ if !inString && !inIdentifier {
+ squareDepth++
+ }
+ case ']':
+ if !inString && !inIdentifier {
+ squareDepth--
+ }
+ case '{':
+ if !inString && !inIdentifier {
+ braceDepth++
+ }
+ case '}':
+ if !inString && !inIdentifier {
+ braceDepth--
+ }
+ }
+ if angleDepth < 0 || parenDepth < 0 || squareDepth < 0 ||
braceDepth < 0 {
+ return nil, errors.New("unbalanced delimiters")
+ }
+ if character == separator && !inString && !inIdentifier &&
angleDepth == 0 && parenDepth == 0 && squareDepth == 0 && braceDepth == 0 {
+ delete(defaultAtDepth, angleDepth)
+ parts = append(parts,
strings.TrimSpace(input[start:index]))
+ start = index + 1
+ }
+ }
+ if inString || inIdentifier || angleDepth != 0 || parenDepth != 0 ||
squareDepth != 0 || braceDepth != 0 {
+ return nil, errors.New("unbalanced delimiters or quotes")
+ }
+ parts = append(parts, strings.TrimSpace(input[start:]))
+
+ return parts, nil
+}
+
+func parseRowField(input string) (string, string, *string, *string, error) {
+ trimmed := strings.TrimSpace(input)
+ if trimmed == "" {
+ return "", "", nil, nil, errors.New("empty field declaration")
+ }
+
+ name, remainder, err := consumeRowFieldName(trimmed)
+ if err != nil {
+ return "", "", nil, nil, err
+ }
+ if strings.TrimSpace(remainder) == "" {
+ return "", "", nil, nil, fmt.Errorf("field %q is missing a
type", name)
+ }
+
+ commentPosition := findTopLevelKeyword(remainder, "COMMENT")
+ defaultPosition := findTopLevelKeyword(remainder, "DEFAULT")
+ typeEnd := len(remainder)
+ if commentPosition >= 0 && commentPosition < typeEnd {
+ typeEnd = commentPosition
+ }
+ if defaultPosition >= 0 && defaultPosition < typeEnd {
+ typeEnd = defaultPosition
+ }
+ fieldType := strings.TrimSpace(remainder[:typeEnd])
+ if fieldType == "" {
+ return "", "", nil, nil, fmt.Errorf("field %q is missing a
type", name)
+ }
+
+ var description, defaultValue *string
+ if commentPosition >= 0 {
+ if defaultPosition >= 0 && defaultPosition < commentPosition {
+ return "", "", nil, nil, fmt.Errorf("field %q COMMENT
must precede DEFAULT", name)
+ }
+ commentEnd := len(remainder)
+ if defaultPosition >= 0 {
+ commentEnd = defaultPosition
+ }
+ commentText :=
strings.TrimSpace(remainder[commentPosition+len("COMMENT") : commentEnd])
+ decoded, err := decodeSQLString(commentText)
+ if err != nil {
+ return "", "", nil, nil, fmt.Errorf("field %q has
invalid COMMENT: %w", name, err)
+ }
+ description = &decoded
+ }
+ if defaultPosition >= 0 {
+ value :=
strings.TrimSpace(remainder[defaultPosition+len("DEFAULT"):])
+ if value == "" {
+ return "", "", nil, nil, fmt.Errorf("field %q has an
empty DEFAULT", name)
+ }
+ defaultValue = &value
+ }
+
+ return name, fieldType, description, defaultValue, nil
+}
+
+func consumeRowFieldName(input string) (string, string, error) {
+ if input[0] != '`' {
+ index := strings.IndexFunc(input, unicode.IsSpace)
+ if index <= 0 {
+ return "", "", errors.New("expected a field name
followed by a type")
+ }
+
+ return input[:index], input[index:], nil
+ }
+
+ var name strings.Builder
+ for index := 1; index < len(input); index++ {
+ if input[index] != '`' {
+ name.WriteByte(input[index])
+
+ continue
+ }
+ if index+1 < len(input) && input[index+1] == '`' {
+ name.WriteByte('`')
+ index++
+
+ continue
+ }
+ if index+1 < len(input) &&
!unicode.IsSpace(rune(input[index+1])) {
+ return "", "", errors.New("expected whitespace after
quoted field name")
+ }
+
+ return name.String(), input[index+1:], nil
+ }
+
+ return "", "", errors.New("unterminated quoted field name")
+}
+
+func findTopLevelKeyword(input, keyword string) int {
+ angleDepth, parenDepth, squareDepth, braceDepth := 0, 0, 0, 0
+ inString := false
+ inIdentifier := false
+ for index := 0; index < len(input); index++ {
+ switch input[index] {
+ case '\'':
+ if inIdentifier {
+ continue
+ }
+ if inString && index+1 < len(input) && input[index+1]
== '\'' {
+ index++
+
+ continue
+ }
+ inString = !inString
+ case '`':
+ if inString {
+ continue
+ }
+ if inIdentifier && index+1 < len(input) &&
input[index+1] == '`' {
+ index++
+
+ continue
+ }
+ inIdentifier = !inIdentifier
+ case '<':
+ if !inString && !inIdentifier && parenDepth == 0 &&
squareDepth == 0 && braceDepth == 0 && isCompositeAngleOpening(input, index) {
+ angleDepth++
+ }
+ case '>':
+ if !inString && !inIdentifier && parenDepth == 0 &&
squareDepth == 0 && braceDepth == 0 && isStructuralAngleClosing(input, index) {
+ angleDepth--
+ }
+ case '(':
+ if !inString && !inIdentifier {
+ parenDepth++
+ }
+ case ')':
+ if !inString && !inIdentifier {
+ parenDepth--
+ }
+ case '[':
+ if !inString && !inIdentifier {
+ squareDepth++
+ }
+ case ']':
+ if !inString && !inIdentifier {
+ squareDepth--
+ }
+ case '{':
+ if !inString && !inIdentifier {
+ braceDepth++
+ }
+ case '}':
+ if !inString && !inIdentifier {
+ braceDepth--
+ }
+ default:
+ if !inString && !inIdentifier && angleDepth == 0 &&
parenDepth == 0 && squareDepth == 0 && braceDepth == 0 && hasKeywordAt(input,
index, keyword) {
+ return index
+ }
+ }
+ }
+
+ return -1
+}
+
+func isCompositeAngleOpening(input string, index int) bool {
+ if index < 0 || index >= len(input) || input[index] != '<' {
+ return false
+ }
+ end := index
+ for end > 0 && unicode.IsSpace(rune(input[end-1])) {
+ end--
+ }
+ start := end
+ for start > 0 {
+ character := input[start-1]
+ if character != '_' && (character < 'A' || character > 'Z') &&
(character < 'a' || character > 'z') {
+ break
+ }
+ start--
+ }
+
+ switch strings.ToUpper(input[start:end]) {
+ case "ARRAY", "MAP", "MULTISET", "ROW", "VECTOR":
+ return true
+ default:
+ return false
+ }
+}
+
+func isStructuralAngleClosing(input string, index int) bool {
+ if index < 0 || index >= len(input) || input[index] != '>' {
+ return false
+ }
+ next := index + 1
+ for next < len(input) && unicode.IsSpace(rune(input[next])) {
+ next++
+ }
+ if next == len(input) {
+ return true
+ }
+ switch input[next] {
+ case ',', '>', ')', ']', '}':
+ return true
+ }
+
+ return hasKeywordAt(input, next, "NOT") || hasKeywordAt(input, next,
"COMMENT") || hasKeywordAt(input, next, "DEFAULT")
+}
+
+func hasKeywordAt(input string, index int, keyword string) bool {
+ if index+len(keyword) > len(input) ||
!strings.EqualFold(input[index:index+len(keyword)], keyword) {
+ return false
+ }
+ if index > 0 && !unicode.IsSpace(rune(input[index-1])) {
+ return false
+ }
+ end := index + len(keyword)
+
+ return end == len(input) || unicode.IsSpace(rune(input[end]))
+}
+
+func decodeSQLString(input string) (string, error) {
+ trimmed := strings.TrimSpace(input)
+ if len(trimmed) < 2 || trimmed[0] != '\'' || trimmed[len(trimmed)-1] !=
'\'' {
+ return "", errors.New("expected a single-quoted string")
+ }
+ contents := trimmed[1 : len(trimmed)-1]
+ for index := 0; index < len(contents); index++ {
+ if contents[index] != '\'' {
+ continue
+ }
+ if index+1 >= len(contents) || contents[index+1] != '\'' {
+ return "", errors.New("unescaped single quote")
+ }
+ index++
+ }
+
+ return strings.ReplaceAll(contents, "''", "'"), nil
+}
diff --git a/internal/client/dlf_auth.go b/internal/client/dlf_auth.go
index d9f6391..dc44bb5 100644
--- a/internal/client/dlf_auth.go
+++ b/internal/client/dlf_auth.go
@@ -145,6 +145,7 @@ func newDLFAuthenticator(endpoint *url.URL, config
DLFConfig, defaultHTTPClient
if httpClient == nil {
httpClient = defaultHTTPClient
}
+ httpClient = noRedirectHTTPClient(httpClient)
provider, err := newDLFCredentialProvider(config, httpClient)
if err != nil {
return nil, err
@@ -343,21 +344,15 @@ func (l *fileDLFTokenLoader) Load(ctx context.Context)
(dlfCredentials, error) {
}
var lastErr error
for attempt := 1; attempt <= attempts; attempt++ {
- contents, err := os.ReadFile(l.path)
+ contents, err := readDLFTokenFile(l.path)
if err == nil {
- if len(contents) > 1<<20 {
- err = errors.New("DLF token file is larger than
1 MiB")
- } else {
- var credentials dlfCredentials
- if decodeErr := json.Unmarshal(contents,
&credentials); decodeErr == nil {
- if validationErr :=
credentials.validate(); validationErr == nil {
- return credentials, nil
- }
+ var credentials dlfCredentials
+ if decodeErr := json.Unmarshal(contents, &credentials);
decodeErr == nil {
+ if validationErr := credentials.validate();
validationErr == nil {
+ return credentials, nil
}
- err = errors.New("failed to parse DLF token
file")
}
- } else {
- err = errors.New("failed to read DLF token file")
+ err = errors.New("failed to parse DLF token file")
}
lastErr = err
if attempt == attempts {
@@ -377,6 +372,24 @@ func (l *fileDLFTokenLoader) Load(ctx context.Context)
(dlfCredentials, error) {
return dlfCredentials{}, lastErr
}
+func readDLFTokenFile(path string) ([]byte, error) {
+ file, err := os.Open(path)
+ if err != nil {
+ return nil, errors.New("failed to read DLF token file")
+ }
+ defer file.Close()
+
+ contents, err := io.ReadAll(io.LimitReader(file, (1<<20)+1))
+ if err != nil {
+ return nil, errors.New("failed to read DLF token file")
+ }
+ if len(contents) > 1<<20 {
+ return nil, errors.New("DLF token file is larger than 1 MiB")
+ }
+
+ return contents, nil
+}
+
type ecsDLFTokenLoader struct {
metadataURL string
roleName string
diff --git a/internal/client/dlf_auth_test.go b/internal/client/dlf_auth_test.go
index d7678e2..d1147e2 100644
--- a/internal/client/dlf_auth_test.go
+++ b/internal/client/dlf_auth_test.go
@@ -194,6 +194,15 @@ func
TestFileDLFTokenLoaderLoadsSTSAndDoesNotLeakMalformedContent(t *testing.T)
assert.NotContains(t, err.Error(), "must-not-appear-in-error")
}
+func TestFileDLFTokenLoaderRejectsOversizedFile(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "token.json")
+ require.NoError(t, os.WriteFile(path, []byte(strings.Repeat("x",
(1<<20)+1)), 0o600))
+
+ loader := &fileDLFTokenLoader{path: path, maxAttempts: 1}
+ _, err := loader.Load(context.Background())
+ require.EqualError(t, err, "DLF token file is larger than 1 MiB")
+}
+
func TestECSDLFTokenLoaderDiscoversAndCachesRole(t *testing.T) {
var roleRequests atomic.Int32
var credentialRequests atomic.Int32
diff --git a/internal/client/models.go b/internal/client/models.go
index cfbc691..0e72876 100644
--- a/internal/client/models.go
+++ b/internal/client/models.go
@@ -101,12 +101,18 @@ type alterTableRequest struct {
}
// DataType is Paimon's language-neutral REST representation of a data type.
-// It is encoded as Paimon's SQL type string on writes. Reads also accept the
-// structured JSON form used for ARRAY, MAP, MULTISET, ROW and VECTOR values.
+// Atomic types use SQL strings. ARRAY, MAP, MULTISET, ROW and VECTOR use the
+// structured JSON form required by Paimon's REST type parser.
type DataType string
func (t DataType) MarshalJSON() ([]byte, error) {
- return json.Marshal(string(t))
+ nextFieldID := -1
+ value, err := encodeDataType(string(t), &nextFieldID)
+ if err != nil {
+ return nil, err
+ }
+
+ return json.Marshal(value)
}
func (t *DataType) UnmarshalJSON(data []byte) error {
@@ -155,7 +161,7 @@ func (t *DataType) UnmarshalJSON(data []byte) error {
fields := make([]string, 0, len(structured.Fields))
for _, field := range structured.Fields {
part := quoteIdentifier(field.Name) + " " +
string(field.Type)
- if field.Description != nil {
+ if field.Description != nil && *field.Description != ""
{
part += " COMMENT '" +
strings.ReplaceAll(*field.Description, "'", "''") + "'"
}
if field.DefaultValue != nil {
diff --git a/internal/provider/provider_test.go
b/internal/provider/provider_test.go
index 287ca87..d925e0b 100644
--- a/internal/provider/provider_test.go
+++ b/internal/provider/provider_test.go
@@ -19,13 +19,21 @@ package provider
import (
"context"
+ "encoding/json"
+ "net/http"
+ "net/http/httptest"
"testing"
"github.com/apache/terraform-provider-paimon/internal/client"
+ "github.com/hashicorp/terraform-plugin-framework/attr"
"github.com/hashicorp/terraform-plugin-framework/datasource"
+ "github.com/hashicorp/terraform-plugin-framework/diag"
+ "github.com/hashicorp/terraform-plugin-framework/path"
frameworkprovider
"github.com/hashicorp/terraform-plugin-framework/provider"
providerschema
"github.com/hashicorp/terraform-plugin-framework/provider/schema"
"github.com/hashicorp/terraform-plugin-framework/resource"
+ "github.com/hashicorp/terraform-plugin-framework/schema/validator"
+ "github.com/hashicorp/terraform-plugin-framework/tfsdk"
"github.com/hashicorp/terraform-plugin-framework/types"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -123,3 +131,328 @@ func
TestSchemaFromResourceModelNormalizesPrimaryKeyNullability(t *testing.T) {
require.Len(t, tableSchema.Fields, 1)
assert.Equal(t, client.DataType("BIGINT NOT NULL"),
tableSchema.Fields[0].Type)
}
+
+func TestSchemaFromResourceModelAllocatesUnusedFieldIDs(t *testing.T) {
+ ctx := context.Background()
+ fields, diagnostics := types.ListValueFrom(ctx,
types.ObjectType{AttrTypes: tableFieldAttrTypes()}, []tableFieldModel{
+ tableFieldForTest("first", types.Int64Value(1)),
+ tableFieldForTest("second", types.Int64Unknown()),
+ tableFieldForTest("third", types.Int64Value(3)),
+ tableFieldForTest("fourth", types.Int64Null()),
+ tableFieldForTest("max", types.Int64Value(maxPaimonFieldID)),
+ })
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ model := tableResourceModel{
+ Fields: fields,
+ PartitionKeys: types.ListNull(types.StringType),
+ PrimaryKeys: types.ListNull(types.StringType),
+ Options: types.MapNull(types.StringType),
+ Comment: types.StringNull(),
+ }
+
+ tableSchema := schemaFromResourceModel(ctx, &model, &diagnostics)
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ require.Len(t, tableSchema.Fields, 5)
+ assert.Equal(t, []int{1, 0, 3, 2, maxPaimonFieldID}, []int{
+ tableSchema.Fields[0].ID,
+ tableSchema.Fields[1].ID,
+ tableSchema.Fields[2].ID,
+ tableSchema.Fields[3].ID,
+ tableSchema.Fields[4].ID,
+ })
+}
+
+func TestSchemaFromResourceModelRejectsInvalidFieldIDs(t *testing.T) {
+ ctx := context.Background()
+ fields, diagnostics := types.ListValueFrom(ctx,
types.ObjectType{AttrTypes: tableFieldAttrTypes()}, []tableFieldModel{
+ tableFieldForTest("first", types.Int64Value(2)),
+ tableFieldForTest("duplicate", types.Int64Value(2)),
+ tableFieldForTest("negative", types.Int64Value(-1)),
+ tableFieldForTest("reserved",
types.Int64Value(maxPaimonFieldID+1)),
+ })
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ model := tableResourceModel{
+ Fields: fields,
+ PartitionKeys: types.ListNull(types.StringType),
+ PrimaryKeys: types.ListNull(types.StringType),
+ Options: types.MapNull(types.StringType),
+ Comment: types.StringNull(),
+ }
+
+ _ = schemaFromResourceModel(ctx, &model, &diagnostics)
+ require.True(t, diagnostics.HasError())
+ require.Len(t, diagnostics.Errors(), 3)
+ assert.Contains(t, diagnostics.Errors()[0].Summary(), "Duplicate Paimon
field ID")
+ assert.Contains(t, diagnostics.Errors()[1].Summary(), "Invalid Paimon
field ID")
+ assert.Contains(t, diagnostics.Errors()[2].Summary(), "Invalid Paimon
field ID")
+}
+
+func TestReservedTableOptionsValidator(t *testing.T) {
+ ctx := context.Background()
+ options, diagnostics := types.MapValueFrom(ctx, types.StringType,
map[string]string{
+ "bucket": "4",
+ "partition": "dt",
+ "primary-key": "id",
+ })
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+
+ var response validator.MapResponse
+ reservedTableOptionsValidator{}.ValidateMap(ctx, validator.MapRequest{
+ Path: path.Root("options"),
+ ConfigValue: options,
+ }, &response)
+ require.True(t, response.Diagnostics.HasError())
+ assert.Contains(t, response.Diagnostics.Errors()[0].Detail(),
"partition, primary-key")
+}
+
+func TestImmutableTableOptionsChanged(t *testing.T) {
+ mapValue := func(values map[string]attr.Value) types.Map {
+ return types.MapValueMust(types.StringType, values)
+ }
+
+ assert.False(t, immutableTableOptionsChanged(
+ mapValue(map[string]attr.Value{"bucket":
types.StringValue("2")}),
+ mapValue(map[string]attr.Value{"bucket":
types.StringValue("4")}),
+ ))
+ assert.True(t, immutableTableOptionsChanged(
+ mapValue(map[string]attr.Value{"merge-engine":
types.StringValue("deduplicate")}),
+ mapValue(map[string]attr.Value{"merge-engine":
types.StringValue("partial-update")}),
+ ))
+ assert.True(t, immutableTableOptionsChanged(
+ mapValue(map[string]attr.Value{"primary-key.nullable":
types.StringValue("true")}),
+ mapValue(map[string]attr.Value{}),
+ ))
+ assert.False(t, immutableTableOptionsChanged(
+ types.MapNull(types.StringType),
+ mapValue(map[string]attr.Value{"type":
types.StringValue("table")}),
+ ))
+ assert.False(t, immutableTableOptionsChanged(
+ mapValue(map[string]attr.Value{"type":
types.StringValue("table")}),
+ mapValue(map[string]attr.Value{"type":
types.StringValue("TABLE")}),
+ ))
+ assert.False(t, immutableTableOptionsChanged(
+ mapValue(map[string]attr.Value{"type":
types.StringValue("table")}),
+ mapValue(map[string]attr.Value{}),
+ ))
+ assert.False(t, immutableTableOptionsChanged(
+ types.MapUnknown(types.StringType),
+ mapValue(map[string]attr.Value{"merge-engine":
types.StringValue("partial-update")}),
+ ))
+ assert.False(t, immutableTableOptionsChanged(
+ mapValue(map[string]attr.Value{"merge-engine":
types.StringValue("deduplicate")}),
+ mapValue(map[string]attr.Value{"merge-engine":
types.StringUnknown()}),
+ ))
+ assert.True(t, immutableTableOptionsChanged(
+ mapValue(map[string]attr.Value{
+ "bucket-key": types.StringValue("id"),
+ "merge-engine": types.StringValue("deduplicate"),
+ }),
+ mapValue(map[string]attr.Value{
+ "bucket-key": types.StringUnknown(),
+ "merge-engine": types.StringValue("partial-update"),
+ }),
+ ))
+}
+
+func TestTableTypeSemanticNoOpPreservesConfiguredValue(t *testing.T) {
+ removals, updates := diffTableOptions(
+ map[string]string{},
+ map[string]string{"type": "table"},
+ )
+ assert.Empty(t, removals)
+ assert.Empty(t, updates)
+
+ removals, updates = diffTableOptions(
+ map[string]string{"type": "table"},
+ map[string]string{"type": "TABLE"},
+ )
+ assert.Empty(t, removals)
+ assert.Empty(t, updates)
+
+ removals, updates = diffTableOptions(
+ map[string]string{"type": "table"},
+ map[string]string{},
+ )
+ assert.Empty(t, removals)
+ assert.Empty(t, updates)
+
+ ctx := context.Background()
+ managed := types.MapValueMust(types.StringType, map[string]attr.Value{
+ "type": types.StringValue("TABLE"),
+ })
+ var diagnostics diag.Diagnostics
+ synced := syncManagedTableOptions(ctx, managed, map[string]string{},
&diagnostics)
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ assert.Equal(t, map[string]string{"type": "TABLE"}, mapFromValue(ctx,
synced, &diagnostics))
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+
+ synced = syncManagedTableOptions(ctx, managed,
map[string]string{"type": "table"}, &diagnostics)
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ assert.Equal(t, map[string]string{"type": "TABLE"}, mapFromValue(ctx,
synced, &diagnostics))
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+}
+
+func TestTableResourceLifecycle(t *testing.T) {
+ ctx := context.Background()
+ remote := client.Table{
+ ID: "table-id",
+ Database: "analytics",
+ Name: "events",
+ SchemaID: 1,
+ Schema: client.Schema{
+ Fields: []client.Field{
+ {ID: 0, Name: "id", Type:
client.DataType("BIGINT NOT NULL")},
+ {ID: 1, Name: "labels", Type:
client.DataType("MAP<STRING, STRING>")},
+ {ID: 2, Name: "payload", Type:
client.DataType("ROW<`item` STRING>")},
+ },
+ PartitionKeys: []string{},
+ PrimaryKeys: []string{"id"},
+ Options: map[string]string{"bucket": "2",
"server-only": "preserved"},
+ },
+ }
+ createCalls, readCalls, updateCalls, deleteCalls := 0, 0, 0, 0
+ server := httptest.NewServer(http.HandlerFunc(func(w
http.ResponseWriter, request *http.Request) {
+ w.Header().Set("Content-Type", "application/json")
+ switch {
+ case request.Method == http.MethodGet && request.URL.Path ==
"/v1/config":
+ require.NoError(t,
json.NewEncoder(w).Encode(client.ConfigResponse{Defaults:
map[string]string{"prefix": "catalog"}}))
+ case request.Method == http.MethodPost && request.URL.Path ==
"/v1/catalog/databases/analytics/tables":
+ createCalls++
+ var body struct {
+ Identifier client.Identifier `json:"identifier"`
+ Schema client.Schema `json:"schema"`
+ }
+ require.NoError(t,
json.NewDecoder(request.Body).Decode(&body))
+ assert.Equal(t, "events", body.Identifier.Object)
+ assert.Equal(t, map[string]string{"bucket": "2"},
body.Schema.Options)
+ w.WriteHeader(http.StatusOK)
+ case request.Method == http.MethodGet && request.URL.Path ==
"/v1/catalog/databases/analytics/tables/events":
+ readCalls++
+ require.NoError(t, json.NewEncoder(w).Encode(remote))
+ case request.Method == http.MethodPost && request.URL.Path ==
"/v1/catalog/databases/analytics/tables/events":
+ updateCalls++
+ var body struct {
+ Changes []client.SchemaChange `json:"changes"`
+ }
+ require.NoError(t,
json.NewDecoder(request.Body).Decode(&body))
+ assert.Equal(t, []client.SchemaChange{{"action":
"setOption", "key": "bucket", "value": "4"}}, body.Changes)
+ remote.Schema.Options["bucket"] = "4"
+ remote.SchemaID++
+ w.WriteHeader(http.StatusOK)
+ case request.Method == http.MethodDelete && request.URL.Path ==
"/v1/catalog/databases/analytics/tables/events":
+ deleteCalls++
+ w.WriteHeader(http.StatusNoContent)
+ default:
+ http.NotFound(w, request)
+ }
+ }))
+ defer server.Close()
+
+ api, err := client.New(client.Config{URI: server.URL})
+ require.NoError(t, err)
+ table := &tableResource{client: api}
+ var schemaResponse resource.SchemaResponse
+ table.Schema(ctx, resource.SchemaRequest{}, &schemaResponse)
+ require.False(t, schemaResponse.Diagnostics.HasError(),
schemaResponse.Diagnostics.Errors())
+
+ fields, diagnostics := types.ListValueFrom(ctx,
types.ObjectType{AttrTypes: tableFieldAttrTypes()}, []tableFieldModel{
+ {
+ ID: types.Int64Unknown(),
+ Name: types.StringValue("id"),
+ Type: types.StringValue("BIGINT"),
+ Nullable: types.BoolValue(false),
+ Description: types.StringNull(),
+ DefaultValue: types.StringNull(),
+ },
+ {
+ ID: types.Int64Unknown(),
+ Name: types.StringValue("labels"),
+ Type: types.StringValue("MAP<STRING,STRING>"),
+ Nullable: types.BoolValue(true),
+ Description: types.StringNull(),
+ DefaultValue: types.StringNull(),
+ },
+ {
+ ID: types.Int64Unknown(),
+ Name: types.StringValue("payload"),
+ Type: types.StringValue("ROW<`item` STRING>"),
+ Nullable: types.BoolValue(true),
+ Description: types.StringNull(),
+ DefaultValue: types.StringNull(),
+ },
+ })
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ primaryKeys, diagnostics := types.ListValueFrom(ctx, types.StringType,
[]string{"id"})
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ options, diagnostics := types.MapValueFrom(ctx, types.StringType,
map[string]string{"bucket": "2"})
+ require.False(t, diagnostics.HasError(), diagnostics.Errors())
+ planModel := tableResourceModel{
+ ID: types.StringUnknown(),
+ CatalogID: types.StringUnknown(),
+ Database: types.StringValue("analytics"),
+ Name: types.StringValue("events"),
+ Fields: fields,
+ PartitionKeys: types.ListValueMust(types.StringType,
[]attr.Value{}),
+ PrimaryKeys: primaryKeys,
+ Options: options,
+ ServerOptions: types.MapUnknown(types.StringType),
+ Comment: types.StringNull(),
+ SchemaID: types.Int64Unknown(),
+ Path: types.StringUnknown(),
+ IsExternal: types.BoolUnknown(),
+ Owner: types.StringUnknown(),
+ CreatedAt: types.Int64Unknown(),
+ CreatedBy: types.StringUnknown(),
+ UpdatedAt: types.Int64Unknown(),
+ UpdatedBy: types.StringUnknown(),
+ }
+ plan := tfsdk.Plan{Schema: schemaResponse.Schema}
+ require.False(t, plan.Set(ctx, &planModel).HasError())
+ createResponse := resource.CreateResponse{State: tfsdk.State{Schema:
schemaResponse.Schema}}
+ table.Create(ctx, resource.CreateRequest{Plan: plan}, &createResponse)
+ require.False(t, createResponse.Diagnostics.HasError(),
createResponse.Diagnostics.Errors())
+ var createdModel tableResourceModel
+ require.False(t, createResponse.State.Get(ctx,
&createdModel).HasError())
+ var createdFields []tableFieldModel
+ require.False(t, createdModel.Fields.ElementsAs(ctx, &createdFields,
false).HasError())
+ require.Len(t, createdFields, 3)
+ assert.Equal(t, "MAP<STRING,STRING>",
createdFields[1].Type.ValueString())
+ assert.Equal(t, "ROW<`item` STRING>",
createdFields[2].Type.ValueString())
+
+ readResponse := resource.ReadResponse{State: createResponse.State}
+ table.Read(ctx, resource.ReadRequest{State: createResponse.State},
&readResponse)
+ require.False(t, readResponse.Diagnostics.HasError(),
readResponse.Diagnostics.Errors())
+
+ var updateModel tableResourceModel
+ require.False(t, readResponse.State.Get(ctx, &updateModel).HasError())
+ var refreshedFields []tableFieldModel
+ require.False(t, updateModel.Fields.ElementsAs(ctx, &refreshedFields,
false).HasError())
+ require.Len(t, refreshedFields, 3)
+ assert.Equal(t, "MAP<STRING,STRING>",
refreshedFields[1].Type.ValueString())
+ assert.Equal(t, "ROW<`item` STRING>",
refreshedFields[2].Type.ValueString())
+ updateModel.Options = types.MapValueMust(types.StringType,
map[string]attr.Value{"bucket": types.StringValue("4")})
+ updatePlan := tfsdk.Plan{Schema: schemaResponse.Schema}
+ require.False(t, updatePlan.Set(ctx, &updateModel).HasError())
+ updateResponse := resource.UpdateResponse{State: tfsdk.State{Schema:
schemaResponse.Schema}}
+ table.Update(ctx, resource.UpdateRequest{State: readResponse.State,
Plan: updatePlan}, &updateResponse)
+ require.False(t, updateResponse.Diagnostics.HasError(),
updateResponse.Diagnostics.Errors())
+
+ deleteResponse := resource.DeleteResponse{State: updateResponse.State}
+ table.Delete(ctx, resource.DeleteRequest{State: updateResponse.State},
&deleteResponse)
+ require.False(t, deleteResponse.Diagnostics.HasError(),
deleteResponse.Diagnostics.Errors())
+ assert.Equal(t, 1, createCalls)
+ assert.Equal(t, 3, readCalls)
+ assert.Equal(t, 1, updateCalls)
+ assert.Equal(t, 1, deleteCalls)
+}
+
+func tableFieldForTest(name string, id types.Int64) tableFieldModel {
+ return tableFieldModel{
+ ID: id,
+ Name: types.StringValue(name),
+ Type: types.StringValue("STRING"),
+ Nullable: types.BoolValue(true),
+ Description: types.StringNull(),
+ DefaultValue: types.StringNull(),
+ }
+}
diff --git a/internal/provider/resource_table.go
b/internal/provider/resource_table.go
index 9ad97e6..9714f3d 100644
--- a/internal/provider/resource_table.go
+++ b/internal/provider/resource_table.go
@@ -137,7 +137,7 @@ func (r *tableResource) Update(ctx context.Context, req
resource.UpdateRequest,
if resp.Diagnostics.HasError() {
return
}
- removals, updates := diffOptions(before, after)
+ removals, updates := diffTableOptions(before, after)
sort.Strings(removals)
updateKeys := make([]string, 0, len(updates))
for key := range updates {
@@ -216,10 +216,10 @@ func setTableResourceModel(ctx context.Context, model
*tableResourceModel, table
model.CatalogID = types.StringValue(table.ID)
model.Database = types.StringValue(database)
model.Name = types.StringValue(name)
- model.Fields = fieldsValueFromRemote(ctx, table.Schema.Fields, diags)
+ 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 = syncManagedOptions(ctx, model.Options,
table.Schema.Options, diags)
+ model.Options = syncManagedTableOptions(ctx, model.Options,
table.Schema.Options, diags)
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
new file mode 100644
index 0000000..d03f485
--- /dev/null
+++ b/internal/provider/table_options.go
@@ -0,0 +1,184 @@
+// 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"
+ "strings"
+
+ "github.com/hashicorp/terraform-plugin-framework/diag"
+
"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"
+ "github.com/hashicorp/terraform-plugin-framework/types"
+)
+
+var immutableTableOptions = map[string]struct{}{
+ "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": {},
+ "partition": {},
+ "pk-clustering-override": {},
+ "primary-key": {},
+ "primary-key.nullable": {},
+ "row-tracking.enabled": {},
+ "rowkind.field": {},
+ "sequence.snapshot-ordering": {},
+ "type": {},
+}
+
+type reservedTableOptionsValidator struct{}
+
+func (reservedTableOptionsValidator) Description(context.Context) string {
+ return "must configure primary keys and partitions with their
first-class attributes"
+}
+
+func (reservedTableOptionsValidator) MarkdownDescription(context.Context)
string {
+ return "must configure primary keys and partitions with `primary_keys`
and `partition_keys`"
+}
+
+func (reservedTableOptionsValidator) ValidateMap(_ context.Context, req
validator.MapRequest, resp *validator.MapResponse) {
+ if req.ConfigValue.IsNull() || req.ConfigValue.IsUnknown() {
+ return
+ }
+ reserved := make([]string, 0, 2)
+ for _, key := range []string{"partition", "primary-key"} {
+ if _, exists := req.ConfigValue.Elements()[key]; exists {
+ reserved = append(reserved, key)
+ }
+ }
+ if len(reserved) > 0 {
+ resp.Diagnostics.AddAttributeError(
+ req.Path,
+ "Reserved Paimon table option",
+ "Do not configure "+strings.Join(reserved, ", ")+" in
options. Use partition_keys and primary_keys instead.",
+ )
+ }
+}
+
+func immutableTableOptionsRequiresReplace(_ context.Context, req
planmodifier.MapRequest, resp *mapplanmodifier.RequiresReplaceIfFuncResponse) {
+ resp.RequiresReplace = immutableTableOptionsChanged(req.StateValue,
req.PlanValue)
+}
+
+func immutableTableOptionsChanged(before, after types.Map) bool {
+ for key := range immutableTableOptions {
+ beforeValue, beforeExists, beforeKnown :=
knownTableOption(before, key)
+ afterValue, afterExists, afterKnown := knownTableOption(after,
key)
+ if !beforeKnown || !afterKnown {
+ continue
+ }
+ if key == "type" {
+ if !beforeExists {
+ beforeValue = "table"
+ }
+ if !afterExists {
+ afterValue = "table"
+ }
+ if !strings.EqualFold(beforeValue, afterValue) {
+ return true
+ }
+
+ continue
+ }
+ if beforeExists != afterExists || beforeExists && beforeValue
!= afterValue {
+ return true
+ }
+ }
+
+ return false
+}
+
+func knownTableOption(options types.Map, key string) (string, bool, bool) {
+ if options.IsUnknown() {
+ return "", false, false
+ }
+ if options.IsNull() {
+ return "", false, true
+ }
+ element, exists := options.Elements()[key]
+ if !exists {
+ return "", false, true
+ }
+ value, ok := element.(types.String)
+ if !ok || value.IsNull() || value.IsUnknown() {
+ return "", true, false
+ }
+
+ return value.ValueString(), true, true
+}
+
+func diffTableOptions(before, after map[string]string) ([]string,
map[string]string) {
+ removals, updates := diffOptions(before, after)
+ beforeType, beforeExists := before["type"]
+ if !beforeExists {
+ beforeType = "table"
+ }
+ afterType, afterExists := after["type"]
+ if !afterExists {
+ afterType = "table"
+ }
+ if strings.EqualFold(beforeType, afterType) {
+ delete(updates, "type")
+ for index, key := range removals {
+ if key == "type" {
+ removals = append(removals[:index],
removals[index+1:]...)
+
+ break
+ }
+ }
+ }
+
+ return removals, updates
+}
+
+func syncManagedTableOptions(ctx context.Context, managed types.Map, remote
map[string]string, diags *diag.Diagnostics) types.Map {
+ synced := syncManagedOptions(ctx, managed, remote, diags)
+ if managed.IsNull() || managed.IsUnknown() || diags.HasError() {
+ return synced
+ }
+ managedOptions := mapFromValue(ctx, managed, diags)
+ configuredType, managesType := managedOptions["type"]
+ if !managesType || diags.HasError() {
+ return synced
+ }
+ remoteType, hasRemoteType := remote["type"]
+ if !hasRemoteType {
+ remoteType = "table"
+ }
+ if !strings.EqualFold(configuredType, remoteType) {
+ return synced
+ }
+ syncedOptions := mapFromValue(ctx, synced, diags)
+ if diags.HasError() {
+ return synced
+ }
+ syncedOptions["type"] = configuredType
+
+ return stringMapValue(ctx, syncedOptions, diags)
+}
diff --git a/internal/provider/table_schema.go
b/internal/provider/table_schema.go
index 23d2177..abcd91b 100644
--- a/internal/provider/table_schema.go
+++ b/internal/provider/table_schema.go
@@ -29,6 +29,7 @@ import (
"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/mapplanmodifier"
"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"
@@ -44,6 +45,9 @@ type tableFieldModel struct {
DefaultValue types.String `tfsdk:"default_value"`
}
+// Paimon reserves IDs at and above SpecialFields.SYSTEM_FIELD_ID_START.
+const maxPaimonFieldID = (1 << 30) - 2
+
func tableFieldAttrTypes() map[string]attr.Type {
return map[string]attr.Type{
"id": types.Int64Type,
@@ -93,9 +97,11 @@ func tableResourceAttributes() map[string]rschema.Attribute {
PlanModifiers:
[]planmodifier.List{listplanmodifier.RequiresReplace()},
},
"options": rschema.MapAttribute{
- Description: "Table options managed by Terraform.
Options not declared here are preserved.",
- Optional: true,
- ElementType: types.StringType,
+ Description: "Table options managed by Terraform.
Options not declared here are preserved. Paimon options that are immutable
after creation cause replacement when changed.",
+ Optional: true,
+ ElementType: types.StringType,
+ 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")},
},
"server_options": rschema.MapAttribute{
Description: "All table options returned by the REST
Catalog.",
@@ -117,7 +123,7 @@ func tableResourceAttributes() map[string]rschema.Attribute
{
func tableFieldResourceAttributes() map[string]rschema.Attribute {
return map[string]rschema.Attribute{
"id": rschema.Int64Attribute{
- Description: "Stable field ID. It is assigned by
position when omitted.",
+ Description: "Stable Paimon field ID between 0 and " +
strconv.Itoa(maxPaimonFieldID) + ". The next available ID is assigned when
omitted.",
Optional: true,
Computed: true,
},
@@ -181,15 +187,11 @@ func schemaFromResourceModel(ctx context.Context, model
*tableResourceModel, dia
if diags.HasError() {
return client.Schema{}
}
- if len(primaryKeys) > 0 {
- if _, exists := options["primary-key"]; exists {
- diags.AddError("Conflicting primary key configuration",
"Configure primary_keys or the primary-key table option, not both.")
- }
+ if _, exists := options["primary-key"]; exists {
+ diags.AddError("Reserved table option", "Configure primary_keys
instead of the primary-key table option.")
}
- if len(partitionKeys) > 0 {
- if _, exists := options["partition"]; exists {
- diags.AddError("Conflicting partition configuration",
"Configure partition_keys or the partition table option, not both.")
- }
+ 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 {
@@ -207,13 +209,10 @@ func schemaFromResourceModel(ctx context.Context, model
*tableResourceModel, dia
var fieldModels []tableFieldModel
diags.Append(model.Fields.ElementsAs(ctx, &fieldModels, false)...)
+ fieldIDs := allocateFieldIDs(fieldModels, diags)
fields := make([]client.Field, 0, len(fieldModels))
fieldNames := make(map[string]struct{}, len(fieldModels))
for index, field := range fieldModels {
- fieldID := index
- if !field.ID.IsNull() && !field.ID.IsUnknown() {
- fieldID = int(field.ID.ValueInt64())
- }
typeName := strings.TrimSpace(field.Type.ValueString())
hasNotNullSuffix :=
strings.HasSuffix(strings.ToUpper(typeName), " NOT NULL")
_, isPrimaryKey := primaryKeySet[field.Name.ValueString()]
@@ -242,7 +241,7 @@ func schemaFromResourceModel(ctx context.Context, model
*tableResourceModel, dia
typeName += " NOT NULL"
}
fields = append(fields, client.Field{
- ID: fieldID,
+ ID: fieldIDs[index],
Name: field.Name.ValueString(),
Type: client.DataType(typeName),
Description: optionalStringPointer(field.Description),
@@ -265,6 +264,48 @@ func schemaFromResourceModel(ctx context.Context, model
*tableResourceModel, dia
}
}
+func allocateFieldIDs(fields []tableFieldModel, diags *diag.Diagnostics) []int
{
+ ids := make([]int, len(fields))
+ used := make(map[int]int, len(fields))
+ for index, field := range fields {
+ if field.ID.IsNull() || field.ID.IsUnknown() {
+ continue
+ }
+ configured := field.ID.ValueInt64()
+ if configured < 0 || configured > maxPaimonFieldID {
+ diags.AddError("Invalid Paimon field ID", "Field
"+field.Name.ValueString()+" must have a field ID between 0 and
"+strconv.Itoa(maxPaimonFieldID)+".")
+
+ continue
+ }
+ id := int(configured)
+ if previous, duplicate := used[id]; duplicate {
+ diags.AddError("Duplicate Paimon field ID", "Fields
"+fields[previous].Name.ValueString()+" and "+field.Name.ValueString()+" use
the same field ID: "+strconv.Itoa(id))
+
+ continue
+ }
+ ids[index] = id
+ used[id] = index
+ }
+
+ next := 0
+ for index, field := range fields {
+ if !field.ID.IsNull() && !field.ID.IsUnknown() {
+ continue
+ }
+ for {
+ if _, exists := used[next]; !exists {
+ break
+ }
+ next++
+ }
+ ids[index] = next
+ used[next] = index
+ next++
+ }
+
+ return ids
+}
+
func validateKeyFields(attribute string, keys []string, fields
map[string]struct{}, diags *diag.Diagnostics) {
seen := make(map[string]struct{}, len(keys))
for _, key := range keys {
@@ -279,6 +320,38 @@ func validateKeyFields(attribute string, keys []string,
fields map[string]struct
}
func fieldsValueFromRemote(ctx context.Context, fields []client.Field, diags
*diag.Diagnostics) types.List {
+ return fieldsValueFromModels(ctx, fieldModelsFromRemote(fields), diags)
+}
+
+func resourceFieldsValueFromRemote(ctx context.Context, managed types.List,
fields []client.Field, diags *diag.Diagnostics) types.List {
+ models := fieldModelsFromRemote(fields)
+ if managed.IsNull() || managed.IsUnknown() {
+ return fieldsValueFromModels(ctx, models, diags)
+ }
+ var managedModels []tableFieldModel
+ newDiags := managed.ElementsAs(ctx, &managedModels, false)
+ if newDiags.HasError() || len(managedModels) != len(models) {
+ return fieldsValueFromModels(ctx, models, diags)
+ }
+ for index := range models {
+ if managedModels[index].Name.IsNull() ||
managedModels[index].Name.IsUnknown() || managedModels[index].Type.IsNull() ||
managedModels[index].Type.IsUnknown() {
+ continue
+ }
+ 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()),
+ ) {
+ models[index].Type = managedModels[index].Type
+ }
+ }
+
+ return fieldsValueFromModels(ctx, models, diags)
+}
+
+func fieldModelsFromRemote(fields []client.Field) []tableFieldModel {
models := make([]tableFieldModel, 0, len(fields))
for _, field := range fields {
typeName := strings.TrimSpace(string(field.Type))
@@ -296,6 +369,11 @@ func fieldsValueFromRemote(ctx context.Context, fields
[]client.Field, diags *di
DefaultValue:
stringValueFromPointer(field.DefaultValue),
})
}
+
+ return models
+}
+
+func fieldsValueFromModels(ctx context.Context, models []tableFieldModel,
diags *diag.Diagnostics) types.List {
value, newDiags := types.ListValueFrom(ctx, types.ObjectType{AttrTypes:
tableFieldAttrTypes()}, models)
diags.Append(newDiags...)
