diff --git a/.github/workflows/aws-platform-tests.yml b/.github/workflows/aws-platform-tests.yml index d53b285..b01d2f7 100644 --- a/.github/workflows/aws-platform-tests.yml +++ b/.github/workflows/aws-platform-tests.yml @@ -8,8 +8,8 @@ name: AWS Platform Tests # reviewer), and the OIDC trust policy is scoped to that environment so a token # minted anywhere else cannot assume the role. # -# Only EMR Serverless is implemented. The engine input exists so Athena and Glue -# can be added without restructuring; they fail fast until then. +# EMR Serverless, Glue and Redshift Serverless are implemented. The engine input +# exists so Athena can be added without restructuring; it fails fast until then. on: workflow_dispatch: @@ -20,8 +20,17 @@ on: default: emr-serverless options: - emr-serverless + - glue + - redshift - athena # not implemented - - glue # not implemented + redshift-fixtures: + description: > + Build the Spark fixtures first (engine redshift only). Redshift cannot + create format-version 3 tables, branches or tags, so without these the + affected cells can only report "cannot write" and must leave "cannot + read" unmeasured. Costs one short EMR Serverless job per mode. + type: boolean + default: true modes: description: Storage mode(s) type: choice @@ -78,6 +87,7 @@ jobs: AWS_DATA_BUCKET: ${{ secrets.AWS_DATA_BUCKET }} AWS_EMR_JOB_ROLE_ARN: ${{ secrets.AWS_EMR_JOB_ROLE_ARN }} AWS_TABLE_BUCKET_ARN: ${{ secrets.AWS_TABLE_BUCKET_ARN }} + REDSHIFT_ROLE_ARN: ${{ secrets.AWS_REDSHIFT_ROLE_ARN }} RESOURCE_PREFIX: icebergmatrix ENGINE: emr @@ -85,7 +95,7 @@ jobs: - name: Reject unimplemented engines if: inputs.engine == 'athena' run: | - echo "::error::'${{ inputs.engine }}' is not implemented yet; emr-serverless and glue are." + echo "::error::'${{ inputs.engine }}' is not implemented yet; emr-serverless, glue and redshift are." exit 1 - uses: actions/checkout@v7 @@ -133,8 +143,35 @@ jobs: GLUE_VERSION: ${{ inputs.glue-version }} run: uv run --with boto3 python tests/aws/run_glue.py + # Separate step from the suite so a fixture failure is attributable, and so + # the suite still runs without them: the cells that need a fixture then + # report read support as unmeasured rather than guessing. + - name: Build Spark fixtures for Redshift + id: redshift-fixtures + if: '!inputs.dry-run && inputs.engine == ''redshift'' && inputs.redshift-fixtures' + env: + MODES: ${{ inputs.modes }} + EMR_RELEASE_LABEL: ${{ inputs.release-label }} + run: uv run --with boto3 python tests/aws/run_redshift_fixtures.py + continue-on-error: true + + - name: Run feature suite on Redshift Serverless + if: '!inputs.dry-run && inputs.engine == ''redshift''' + env: + MODES: ${{ inputs.modes }} + # Only claim the fixtures exist if the job that builds them succeeded. + # A stale name here would make every fixture-backed cell report a + # missing table instead of an honest "not measured". + REDSHIFT_FIXTURE_DB: ${{ steps.redshift-fixtures.outcome == 'success' && 'icebergmatrix_rsfix' || '' }} + run: uv run --with boto3 python tests/aws/run_redshift.py + # ENGINE tells teardown which S3 prefix to clear; the Glue and S3 Tables # cleanup is prefix-scoped and runs regardless. + # + # Redshift maps to 'emr' deliberately: its fixtures are built by an EMR + # Serverless job and therefore live under emr/, including the EMR + # application that has to be deleted. The suite's own redshift/ prefix is + # swept unconditionally by teardown.py. - name: Tear down billable resources if: always() && !inputs.dry-run env: diff --git a/.github/workflows/flink-tests.yml b/.github/workflows/flink-tests.yml index d4c08fa..82f7f53 100644 --- a/.github/workflows/flink-tests.yml +++ b/.github/workflows/flink-tests.yml @@ -63,6 +63,7 @@ jobs: export REPO_ROOT="${GITHUB_WORKSPACE}" export REPORT_DIR="${GITHUB_WORKSPACE}/test-reports" export FLINK_ICEBERG_VERSION="${ICEBERG_VERSION}" + export PLATFORM_LABEL="Apache Flink ${FLINK_VERSION} (OSS) + Iceberg ${ICEBERG_VERSION}" python tests/flink_feature_tests.py continue-on-error: true diff --git a/infra/aws/README.MD b/infra/aws/README.MD index 47b39ee..cb93483 100644 --- a/infra/aws/README.MD +++ b/infra/aws/README.MD @@ -82,10 +82,19 @@ any other context cannot assume the role. |---|---| | `AWS_CI_ROLE_ARN` | `CiRoleArn` output | | `AWS_EMR_JOB_ROLE_ARN` | `EmrJobRoleArn` output | +| `AWS_REDSHIFT_ROLE_ARN` | `RedshiftRoleArn` output (needed for `engine: redshift`) | | `AWS_DATA_BUCKET` | your S3 bucket name | | `AWS_TABLE_BUCKET_ARN` | your S3 Tables bucket ARN | | `AWS_REGION` | the region everything lives in | +There is deliberately no secret for the Redshift admin password. The namespace is +created with `ManageAdminPassword`, so Secrets Manager owns the credential and the +driver looks the ARN up at run time with `redshift-serverless get-namespace`. +CloudFormation cannot return it: `AWS::RedshiftServerless::Namespace` rejects +`!GetAtt Namespace.AdminPasswordSecretArn` as "must be a readonly property in +schema", which is why the stack exposes `RedshiftAdminSecretHint` (the CLI command +to look it up) rather than the value. + ## 6. Run it Dispatch **AWS Platform Tests** with `dry-run: true` first. That assumes the @@ -192,3 +201,76 @@ create/insert/select/drop. Reading the configuration alone is misleading, becaus the configuration can be exactly right while the resolved class is something else entirely. Note that the probe must use a namespace carrying the resource prefix; any other name fails on IAM and hides the real result. + +## 8. How the Redshift engine differs + +`engine: redshift` runs `tests/redshift_feature_tests.py` through the Redshift Data +API. There is no cluster-side bundle: the suite executes in the runner and every +statement is a `redshift-data` call, which is why `tests/aws/run_redshift.py` is so +much shorter than the EMR and Glue drivers. + +### Four things that cost real time to discover + +- **The auto-mounted catalog cannot write.** `CREATE TABLE awsdatacatalog.. + ... USING ICEBERG` fails with `No session credential found`. Data access through + `awsdatacatalog` is authorised with the caller's IAM session, and a Data API + connection authenticated as a database user has none. Writes need an external + schema that names the role: + + ```sql + CREATE EXTERNAL SCHEMA s FROM DATA CATALOG DATABASE 'glue_db' + IAM_ROLE 'arn:aws:iam:::role/icebergmatrix-redshift'; + CREATE TABLE s.t (id BIGINT, name VARCHAR) USING ICEBERG + LOCATION 's3:////'; + ``` + +- **S3 Tables needs a Glue resource link first.** A table bucket is a *federated* + Glue catalog and Redshift cannot name one directly: both + `"@s3tablescatalog".ns.t` and putting the federated path in `CATALOG_ID` + fail (the latter silently resolves against the default catalog, so the schema is + created and every `SELECT` then raises `EntityNotFoundException`). The working + route is a resource link in the **default** catalog whose `TargetDatabase` points + at the federated namespace, named by an external schema with `CATALOG_ID` set to + the **plain account id**: + + ```bash + aws glue create-database --cli-input-json '{ + "CatalogId": "", + "DatabaseInput": {"Name": "link_name", "TargetDatabase": { + "CatalogId": ":s3tablescatalog/", "DatabaseName": ""}}}' + ``` + ```sql + CREATE EXTERNAL SCHEMA s FROM DATA CATALOG DATABASE 'link_name' + IAM_ROLE '' REGION '' CATALOG_ID ''; + ``` + + With that in place Redshift creates, reads, updates and deletes Iceberg tables in + S3 Tables, omitting `LOCATION` because the table bucket owns placement. + +- **`DROP TABLE` leaks data in the s3buckets mode.** It removes the Glue entry and + leaves the Parquet and metadata behind, so the suite deletes its own + `redshift//` prefix on the way out. In the s3tables mode `DROP TABLE` really + does remove the table, so the two modes need different teardown. + +- **Some features need a table Redshift cannot create.** It refuses + `format-version 3` outright, so `tests/aws/redshift_fixtures.py` builds v3 tables, + deletion vectors, a variant column and a branch/tag with Spark on EMR, and the + suite then tries to read and write them. That is what separates "cannot write" from + "cannot read": Redshift turns out to read v3 tables and apply deletion vectors + correctly while refusing every v3 write, which is partial support rather than none. + Spark 4 on `emr-spark-8.0.0` rejects `GEOMETRY` and `TIMESTAMP_NS` itself, so those + two cells stay unmeasured rather than being guessed. + +### Syntax notes + +- `VARCHAR(N)` is rejected in an Iceberg table ("Use VARCHAR for strings"); use bare + `VARCHAR`. +- The only writable table properties are `format-version` and `compression_type`. + Anything else, including `write.delete.mode` and the bloom-filter properties, is + refused with `... cannot be used in the PROPERTIES clause of "iceberg" table`. +- Defaults from `SHOW TABLE` differ per mode: `compression_type` is `snappy` on S3 + buckets and `zstd` on S3 Tables. +- On S3 Tables, `PARTITIONED BY` on `CREATE TABLE` is accepted and then silently + discarded; `ALTER TABLE ... ADD PARTITION FIELD` afterwards does apply. Always + confirm a partition spec with `SHOW TABLE` rather than trusting the absence of an + error. diff --git a/infra/aws/aws.yaml b/infra/aws/aws.yaml index 7f38f85..788bd3c 100644 --- a/infra/aws/aws.yaml +++ b/infra/aws/aws.yaml @@ -41,10 +41,37 @@ Parameters: Type: String Default: '' Description: Optional email for budget alerts at 80% of the limit + RedshiftBaseCapacity: + Type: Number + Default: 8 + Description: > + Redshift Serverless base capacity in RPUs. 8 is the global minimum; 4 is + only accepted in the regions listed in the 4-RPU launch announcements, + which do not include us-east-1. Serverless bills per RPU-second while a + query runs and nothing while idle, so this is a ceiling on burst cost + rather than a running charge. + RedshiftSubnetIds: + Type: CommaDelimitedList + Default: '' + Description: > + Three or more subnets in different AZs for the Redshift Serverless + workgroup. Leave empty to let Redshift choose from the default VPC, but + note that not every AZ offers Serverless, so passing subnets explicitly is + the reliable option. + RedshiftSecurityGroupIds: + Type: CommaDelimitedList + Default: '' + Description: > + Security groups for the workgroup. Leave empty to use the VPC default. + Nothing needs inbound access: the tests reach Redshift over the Data API, + which is an AWS API call rather than a database connection. Conditions: CreateOidcProvider: !Equals [!Ref ExistingOidcProviderArn, ''] HasBudgetEmail: !Not [!Equals [!Ref BudgetNotificationEmail, '']] + # CommaDelimitedList cannot be compared directly, so join it back to a string. + HasRedshiftSubnets: !Not [!Equals [!Join ['', !Ref RedshiftSubnetIds], '']] + HasRedshiftSecurityGroups: !Not [!Equals [!Join ['', !Ref RedshiftSecurityGroupIds], '']] Resources: @@ -226,6 +253,157 @@ Resources: Action: lakeformation:GetDataAccess Resource: '*' + # --------------------------------------------------------------------------- + # Redshift role: assumed by Redshift itself, not by GitHub. Redshift is the + # only engine here that both reads and writes Iceberg tables through its own + # service identity, so this role needs catalog write access rather than the + # read-only access a query engine would need. + # --------------------------------------------------------------------------- + RedshiftRole: + Type: AWS::IAM::Role + Properties: + RoleName: !Sub '${ResourcePrefix}-redshift' + Description: Attached to the Redshift Serverless namespace for Iceberg reads and writes + AssumeRolePolicyDocument: + Version: '2012-10-17' + Statement: + # Both principals: provisioned Redshift uses redshift.amazonaws.com and + # Serverless uses redshift-serverless.amazonaws.com. Trusting both keeps + # the role usable if this ever moves to a provisioned cluster. + - Effect: Allow + Principal: + Service: + - redshift.amazonaws.com + - redshift-serverless.amazonaws.com + Action: sts:AssumeRole + Condition: + StringEquals: + 'aws:SourceAccount': !Ref AWS::AccountId + Policies: + + # Writes need DeleteObject as well as Put: MERGE stages intermediate + # files under the table location and garbage-collects them at the end of + # the statement, and it fails outright if that delete is denied. + - PolicyName: s3-data-access + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: + - s3:GetObject + - s3:PutObject + - s3:DeleteObject + Resource: !Sub 'arn:${AWS::Partition}:s3:::${DataBucket}/*' + - Effect: Allow + Action: + - s3:ListBucket + - s3:GetBucketLocation + Resource: !Sub 'arn:${AWS::Partition}:s3:::${DataBucket}' + + # Redshift creates, alters and drops Iceberg tables in the Data Catalog, + # so unlike a pure query engine it needs the write verbs too. + - PolicyName: glue-catalog-access + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: + - glue:GetCatalog + - glue:GetDatabase + - glue:GetDatabases + - glue:CreateDatabase + - glue:DeleteDatabase + - glue:GetTable + - glue:GetTables + - glue:CreateTable + - glue:UpdateTable + - glue:DeleteTable + Resource: + - !Sub 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:catalog' + - !Sub 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:database/${ResourcePrefix}*' + - !Sub 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:table/${ResourcePrefix}*/*' + - !Sub 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:userDefinedFunction/${ResourcePrefix}*/*' + + # S3 Tables mode. Redshift addresses a table bucket as the federated + # catalog "@s3tablescatalog", so it needs both the s3tables API + # and the federated Glue hierarchy, exactly as the Spark engines do. + - PolicyName: s3tables-access + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: s3tables:* + Resource: + - !Ref TableBucketArn + - !Sub '${TableBucketArn}/*' + + - PolicyName: glue-s3tables-federation + PolicyDocument: + Version: '2012-10-17' + Statement: + - Effect: Allow + Action: + - glue:GetCatalog + - glue:GetDatabase + - glue:GetDatabases + - glue:CreateDatabase + - glue:DeleteDatabase + - glue:GetTable + - glue:GetTables + - glue:CreateTable + - glue:UpdateTable + - glue:DeleteTable + Resource: + - !Sub 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:catalog/s3tablescatalog' + - !Sub + - 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:catalog/s3tablescatalog/${TableBucketName}' + - TableBucketName: !Select [1, !Split ['/', !Ref TableBucketArn]] + - !Sub + - 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:database/s3tablescatalog/${TableBucketName}/${ResourcePrefix}*' + - TableBucketName: !Select [1, !Split ['/', !Ref TableBucketArn]] + - !Sub + - 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:table/s3tablescatalog/${TableBucketName}/${ResourcePrefix}*/*' + - TableBucketName: !Select [1, !Split ['/', !Ref TableBucketArn]] + - !Sub + - 'arn:${AWS::Partition}:glue:${AWS::Region}:${AWS::AccountId}:userDefinedFunction/s3tablescatalog/${TableBucketName}/${ResourcePrefix}*/*' + - TableBucketName: !Select [1, !Split ['/', !Ref TableBucketArn]] + - Effect: Allow + Action: lakeformation:GetDataAccess + Resource: '*' + + # --------------------------------------------------------------------------- + # Redshift Serverless namespace and workgroup. + # + # ManageAdminPassword hands the admin credentials to Secrets Manager instead of + # this template, so no password is ever a stack parameter, a stack output or a + # GitHub secret. The tests authenticate to the Data API with that secret's ARN. + # --------------------------------------------------------------------------- + RedshiftNamespace: + Type: AWS::RedshiftServerless::Namespace + Properties: + NamespaceName: !Sub '${ResourcePrefix}-ns' + DbName: dev + AdminUsername: icebergmatrix_admin + ManageAdminPassword: true + DefaultIamRoleArn: !GetAtt RedshiftRole.Arn + IamRoles: + - !GetAtt RedshiftRole.Arn + + RedshiftWorkgroup: + Type: AWS::RedshiftServerless::Workgroup + Properties: + WorkgroupName: !Sub '${ResourcePrefix}-wg' + NamespaceName: !Ref RedshiftNamespace + BaseCapacity: !Ref RedshiftBaseCapacity + # No public endpoint. The Data API is reached as an AWS API call, so + # nothing has to connect to the database over the network. + PubliclyAccessible: false + SubnetIds: !If [HasRedshiftSubnets, !Ref RedshiftSubnetIds, !Ref 'AWS::NoValue'] + SecurityGroupIds: !If + - HasRedshiftSecurityGroups + - !Ref RedshiftSecurityGroupIds + - !Ref 'AWS::NoValue' + # --------------------------------------------------------------------------- # CI role: assumed by GitHub Actions via OIDC. Drives EMR Serverless and # cleans up afterwards. It never touches table data directly. @@ -325,6 +503,52 @@ Resources: StringEquals: 'iam:AWSServiceName': ops.emr-serverless.amazonaws.com + # Redshift is driven entirely through the Data API: statements are + # submitted as AWS API calls, so there is no database connection, no + # password and no network path to arrange. The workgroup itself is + # long-lived stack infrastructure, so the CI role only uses it -- it + # cannot create or delete the workgroup or the namespace. + - PolicyName: drive-redshift + PolicyDocument: + Version: '2012-10-17' + Statement: + # The Data API actions take no resource qualifier: authorisation is + # on the workgroup and the credentials, both scoped below. + - Effect: Allow + Action: + - redshift-data:ExecuteStatement + - redshift-data:BatchExecuteStatement + - redshift-data:DescribeStatement + - redshift-data:GetStatementResult + - redshift-data:CancelStatement + - redshift-data:ListStatements + Resource: '*' + - Effect: Allow + Action: + - redshift-serverless:GetWorkgroup + - redshift-serverless:GetNamespace + - redshift-serverless:ListWorkgroups + - redshift-serverless:ListNamespaces + - redshift-serverless:GetCredentials + Resource: + - !Sub 'arn:${AWS::Partition}:redshift-serverless:${AWS::Region}:${AWS::AccountId}:workgroup/*' + - !Sub 'arn:${AWS::Partition}:redshift-serverless:${AWS::Region}:${AWS::AccountId}:namespace/*' + # Read the Secrets Manager secret Redshift created for the admin + # user, so the Data API can authenticate without a stored password. + # + # Scoped by name rather than by exact ARN: CloudFormation does not + # expose the secret ARN as an attribute of the namespace resource + # (GetAtt Namespace.AdminPasswordSecretArn is rejected as not a + # readonly property), and Secrets Manager appends a random suffix + # to every ARN in any case. The "redshift!" prefix is reserved for + # secrets the Redshift service creates, so this cannot widen to a + # secret of ours. + - Effect: Allow + Action: + - secretsmanager:GetSecretValue + - secretsmanager:DescribeSecret + Resource: !Sub 'arn:${AWS::Partition}:secretsmanager:${AWS::Region}:${AWS::AccountId}:secret:redshift!*' + - PolicyName: s3-scripts-and-reports PolicyDocument: Version: '2012-10-17' @@ -414,4 +638,23 @@ Outputs: Value: !Ref TableBucketArn ResourcePrefixOut: Description: Glue databases and S3 Tables namespaces must start with this - Value: !Ref ResourcePrefix \ No newline at end of file + Value: !Ref ResourcePrefix + RedshiftWorkgroupName: + Description: Set as the AWS_REDSHIFT_WORKGROUP environment secret in GitHub + Value: !Ref RedshiftWorkgroup + RedshiftNamespaceName: + Description: Redshift Serverless namespace backing the workgroup + Value: !Ref RedshiftNamespace + RedshiftAdminSecretHint: + Description: > + Redshift creates and manages the admin secret; its ARN is not a + CloudFormation attribute of the namespace, so look it up at runtime with + redshift-serverless get-namespace --namespace-name and read + adminPasswordSecretArn. Nothing here ever holds the password. + Value: !Sub 'aws redshift-serverless get-namespace --namespace-name ${ResourcePrefix}-ns --query namespace.adminPasswordSecretArn' + RedshiftRoleArn: + Description: IAM role attached to the namespace for Iceberg reads and writes + Value: !GetAtt RedshiftRole.Arn + RedshiftDatabase: + Description: Database name inside the namespace + Value: dev \ No newline at end of file diff --git a/src/data/platforms/aws/s3buckets/redshift-s3/redshift-s3.json b/src/data/platforms/aws/s3buckets/redshift-s3/redshift-s3.json index 36a5f80..d7f0902 100644 --- a/src/data/platforms/aws/s3buckets/redshift-s3/redshift-s3.json +++ b/src/data/platforms/aws/s3buckets/redshift-s3/redshift-s3.json @@ -41,9 +41,15 @@ "caveats": [] }, "aws-redshift-s3:copy-on-write:v2": { - "level": "full", - "notes": "Redshift uses copy-on-write for Iceberg table writes", - "caveats": [] + "level": "none", + "notes": "Redshift is merge-on-read only. A DELETE commits position delete files and leaves the data files in place, and copy-on-write cannot be selected: Redshift rejects the Iceberg write.delete.mode and write.update.mode properties outright.", + "caveats": [ + "CREATE TABLE ... TABLE PROPERTIES ('write.delete.mode'='copy-on-write') fails with: \"write.delete.mode\" cannot be used in the PROPERTIES clause of \"iceberg\" table", + "Measured from the Iceberg snapshot summary: after a DELETE, total-position-deletes is 1 and total-data-files is unchanged" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/iceberg-writes.html" + ] }, "aws-redshift-s3:copy-on-write:v3": { "level": "none", @@ -113,9 +119,15 @@ "caveats": [] }, "aws-redshift-s3:time-travel:v2": { - "level": "full", - "notes": "Redshift supports time travel queries on Iceberg tables by timestamp", - "caveats": [] + "level": "none", + "notes": "Redshift exposes only the current state of an Iceberg table. No time-travel syntax is accepted and no snapshot metadata table is queryable, so an older snapshot cannot be read.", + "caveats": [ + "Five spellings were rejected, including FOR SYSTEM_TIME AS OF, FOR TIMESTAMP AS OF and FOR VERSION AS OF: syntax error at or near \"SYSTEM_TIME\"", + "Iceberg metadata tables such as $snapshots and $history are not exposed either" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] }, "aws-redshift-s3:time-travel:v3": { "level": "none", @@ -165,9 +177,16 @@ ] }, "aws-redshift-s3:read-support:v3": { - "level": "none", - "notes": "Redshift does not support Iceberg V3", - "caveats": [] + "level": "partial", + "notes": "Redshift reads format-version 3 tables but cannot write them. A v3 table created by Spark is queried correctly; any INSERT, UPDATE or DELETE against it is refused, and Redshift cannot create a v3 table in the first place.", + "caveats": [ + "Writes fail with: Iceberg version 3 is not supported. Only Iceberg V2 tables are supported for writing", + "CREATE TABLE with 'format-version'='3' is rejected: \"3\" is not a valid value for the \"format-version\" property of \"iceberg\" table", + "Verified by reading a v3 table built by Spark on EMR, since Redshift cannot produce one itself" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] }, "aws-redshift-s3:write-insert:v2": { "level": "full", @@ -375,13 +394,14 @@ "caveats": [] }, "aws-redshift-s3:snowflake-horizon-catalog:v2": { - "level": "unknown", - "notes": "Redshift Spectrum may be able to read Snowflake-managed Iceberg V2 tables via AWS Glue Catalog Federation, similar to Athena. This has not been confirmed. Write access is not expected to be supported.", + "level": "none", + "notes": "Redshift reaches Iceberg only through the AWS Glue Data Catalog. Snowflake Horizon is consumed over Iceberg REST, and this Redshift version has no Iceberg REST client.", "caveats": [ - "Unconfirmed — based on Glue Catalog Federation pattern used by Athena", - "Write operations to external catalogs are not supported by Redshift Spectrum" + "CREATE EXTERNAL SCHEMA ... FROM ICEBERG REST CATALOG is not recognised: it produces the same error as a deliberately meaningless catalog clause, so the clause is discarded rather than understood" ], - "links": [] + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] }, "aws-redshift-s3:snowflake-horizon-catalog:v3": { "level": "none", @@ -396,10 +416,15 @@ "links": [] }, "aws-redshift-s3:deletion-vectors:v3": { - "level": "none", - "notes": "Redshift does not support Iceberg V3, so V3 deletion vectors are unavailable.", - "caveats": [], - "links": [] + "level": "partial", + "notes": "Redshift applies deletion vectors correctly when reading a v3 table, but cannot produce them: it writes only v2 position deletes and refuses to write to a v3 table at all.", + "caveats": [ + "A Spark-built v3 table with rows 1-4 and a deletion vector removing row 2 reads back as exactly 1, 3, 4", + "Writes fail with: Iceberg version 3 is not supported. Only Iceberg V2 tables are supported for writing" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] } } } diff --git a/src/data/platforms/aws/s3tables/redshift-s3/redshift-s3.json b/src/data/platforms/aws/s3tables/redshift-s3/redshift-s3.json index 9b74e44..0949687 100644 --- a/src/data/platforms/aws/s3tables/redshift-s3/redshift-s3.json +++ b/src/data/platforms/aws/s3tables/redshift-s3/redshift-s3.json @@ -41,9 +41,15 @@ "caveats": [] }, "aws-redshift-s3:copy-on-write:v2": { - "level": "full", - "notes": "Copy-on-write supported when writing to S3 Tables via Redshift", - "caveats": [] + "level": "none", + "notes": "Redshift is merge-on-read only. A DELETE commits position delete files and leaves the data files in place, and copy-on-write cannot be selected: Redshift rejects the Iceberg write.delete.mode and write.update.mode properties outright.", + "caveats": [ + "CREATE TABLE ... TABLE PROPERTIES ('write.delete.mode'='copy-on-write') fails with: \"write.delete.mode\" cannot be used in the PROPERTIES clause of \"iceberg\" table", + "Measured from the Iceberg snapshot summary: after a DELETE, total-position-deletes is 1 and total-data-files is unchanged" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/iceberg-writes.html" + ] }, "aws-redshift-s3:copy-on-write:v3": { "level": "none", @@ -93,9 +99,16 @@ "caveats": [] }, "aws-redshift-s3:hidden-partitioning:v2": { - "level": "full", - "notes": "Redshift supports reading hidden-partitioned tables on S3 Tables", - "caveats": [] + "level": "partial", + "notes": "Transform partitioning is reachable on S3 Tables, but not in one statement. PARTITIONED BY on CREATE TABLE is accepted without error and then silently discarded, leaving an unpartitioned table; ALTER TABLE ADD PARTITION FIELD afterwards does apply the transform.", + "caveats": [ + "CREATE TABLE ... PARTITIONED BY (year(ts)) raises no error, but SHOW TABLE then reports no partition spec at all", + "The same statement stores PARTITIONED BY (YEAR(\"ts\")) correctly in the S3 buckets mode, so this is specific to S3 Tables", + "ALTER TABLE ... ADD PARTITION FIELD day(ts) and bucket(8, id) both apply and appear in SHOW TABLE" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-s3Tables.html" + ] }, "aws-redshift-s3:hidden-partitioning:v3": { "level": "none", @@ -123,9 +136,15 @@ "caveats": [] }, "aws-redshift-s3:time-travel:v2": { - "level": "full", - "notes": "Time travel supported via Redshift with S3 Tables", - "caveats": [] + "level": "none", + "notes": "Redshift exposes only the current state of an Iceberg table. No time-travel syntax is accepted and no snapshot metadata table is queryable, so an older snapshot cannot be read.", + "caveats": [ + "Five spellings were rejected, including FOR SYSTEM_TIME AS OF, FOR TIMESTAMP AS OF and FOR VERSION AS OF: syntax error at or near \"SYSTEM_TIME\"", + "Iceberg metadata tables such as $snapshots and $history are not exposed either" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] }, "aws-redshift-s3:time-travel:v3": { "level": "none", @@ -160,9 +179,16 @@ "caveats": [] }, "aws-redshift-s3:read-support:v3": { - "level": "none", - "notes": "Redshift does not support Iceberg V3 with S3 Tables", - "caveats": [] + "level": "partial", + "notes": "Redshift reads format-version 3 tables but cannot write them. A v3 table created by Spark is queried correctly; any INSERT, UPDATE or DELETE against it is refused, and Redshift cannot create a v3 table in the first place.", + "caveats": [ + "Writes fail with: Iceberg version 3 is not supported. Only Iceberg V2 tables are supported for writing", + "CREATE TABLE with 'format-version'='3' is rejected: \"3\" is not a valid value for the \"format-version\" property of \"iceberg\" table", + "Verified by reading a v3 table built by Spark on EMR, since Redshift cannot produce one itself" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] }, "aws-redshift-s3:write-insert:v2": { "level": "full", @@ -205,9 +231,15 @@ "caveats": [] }, "aws-redshift-s3:bloom-filters:v2": { - "level": "unknown", - "notes": "Bloom filter support for S3 Tables via Redshift not explicitly documented", - "caveats": [] + "level": "none", + "notes": "Bloom filters cannot be requested. Redshift accepts only compression_type as a writable Iceberg table property, so the write.parquet.bloom-filter-enabled.* properties cannot be set.", + "caveats": [ + "Rejected with: \"write.parquet.bloom-filter-enabled.column.id\" cannot be used in the PROPERTIES clause of \"iceberg\" table", + "SHOW TABLE reports the writable properties as format-version and compression_type only" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/iceberg-writes.html" + ] }, "aws-redshift-s3:bloom-filters:v3": { "level": "none", @@ -225,9 +257,15 @@ "caveats": [] }, "aws-redshift-s3:aws-glue-catalog:v2": { - "level": "none", - "notes": "S3 Tables uses its own built-in catalog, not AWS Glue", - "caveats": [] + "level": "full", + "notes": "The Glue Data Catalog is how Redshift reaches S3 Tables at all. A table bucket is a federated Glue catalog, and Redshift addresses it through a Glue resource link named by an external schema; tables are then created, written, read and dropped through it.", + "caveats": [ + "A resource link in the default Glue catalog is required: its TargetDatabase points at :s3tablescatalog/, and the external schema names the link with CATALOG_ID set to the plain account id", + "Naming the federated catalog directly does not work: \"@s3tablescatalog\".ns.table is rejected as an unsupported cross-database reference" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-s3Tables.html" + ] }, "aws-redshift-s3:aws-glue-catalog:v3": { "level": "none", @@ -235,9 +273,15 @@ "caveats": [] }, "aws-redshift-s3:rest-catalog:v2": { - "level": "full", - "notes": "S3 Tables exposes a REST-compatible catalog API used by Redshift", - "caveats": [] + "level": "none", + "notes": "Redshift has no Iceberg REST catalog client. S3 Tables does expose an Iceberg REST endpoint, but Redshift cannot consume it: it reaches S3 Tables through the Glue Data Catalog instead.", + "caveats": [ + "CREATE EXTERNAL SCHEMA ... FROM ICEBERG REST CATALOG is not recognised: it produces the same error as a deliberately meaningless catalog clause", + "Redshift tables can be published into Glue's Iceberg REST endpoint, but that is Redshift as a producer, not as a REST client" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-s3Tables.html" + ] }, "aws-redshift-s3:rest-catalog:v3": { "level": "none", @@ -345,13 +389,14 @@ "caveats": [] }, "aws-redshift-s3:snowflake-horizon-catalog:v2": { - "level": "unknown", - "notes": "Redshift Spectrum may be able to read Snowflake-managed Iceberg V2 tables via AWS Glue Catalog Federation, similar to Athena. This has not been confirmed. Write access is not expected to be supported.", + "level": "none", + "notes": "Redshift reaches Iceberg only through the AWS Glue Data Catalog. Snowflake Horizon is consumed over Iceberg REST, and this Redshift version has no Iceberg REST client.", "caveats": [ - "Unconfirmed — based on Glue Catalog Federation pattern used by Athena", - "Write operations to external catalogs are not supported by Redshift Spectrum" + "CREATE EXTERNAL SCHEMA ... FROM ICEBERG REST CATALOG is not recognised: it produces the same error as a deliberately meaningless catalog clause, so the clause is discarded rather than understood" ], - "links": [] + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] }, "aws-redshift-s3:snowflake-horizon-catalog:v3": { "level": "none", @@ -365,9 +410,15 @@ "caveats": [] }, "aws-redshift-s3:deletion-vectors:v3": { - "level": "none", - "notes": "Redshift does not support Iceberg V3, so V3 deletion vectors are unavailable.", - "caveats": [] + "level": "partial", + "notes": "Redshift applies deletion vectors correctly when reading a v3 table, but cannot produce them: it writes only v2 position deletes and refuses to write to a v3 table at all.", + "caveats": [ + "A Spark-built v3 table with rows 1-4 and a deletion vector removing row 2 reads back as exactly 1, 3, 4", + "Writes fail with: Iceberg version 3 is not supported. Only Iceberg V2 tables are supported for writing" + ], + "links": [ + "https://docs.aws.amazon.com/redshift/latest/dg/querying-iceberg.html" + ] } } } diff --git a/tests/aws/platform_common.py b/tests/aws/platform_common.py index 9def5cf..19ff30d 100644 --- a/tests/aws/platform_common.py +++ b/tests/aws/platform_common.py @@ -188,17 +188,22 @@ def summarise(engine: str, title: str, header: list, results: list, reports: dic # Lead with every mode side by side so the outcome is visible without # scrolling past two full matrices. - verdict = ["| Mode | Total | Passed | Failed | Skipped | Errors | Discrepancies |", - "|------|-------|--------|--------|---------|--------|---------------|"] + # Partial is included because otherwise the row does not add up: a suite that + # reports it (Redshift, for read-yes/write-no features) would show + # passed + failed + skipped short of total, which reads like a bug. Engines + # that never report a partial get a 0 and are unaffected. + verdict = ["| Mode | Total | Passed | Partial | Failed | Skipped | Errors | Discrepancies |", + "|------|-------|--------|---------|--------|---------|--------|---------------|"] for r in results: rep = reports.get(r["mode"]) if r["state"] != "SUCCESS" or not rep: state = r["state"] if r["state"] != "SUCCESS" else "NO REPORT" - verdict.append(f"| {r['mode']} | {state} | | | | | |") + verdict.append(f"| {r['mode']} | {state} | | | | | | |") worst = max(worst, 1) continue s = rep["summary"] - verdict.append(f"| {r['mode']} | {s['total']} | {s['passed']} | {s['failed']} | " + verdict.append(f"| {r['mode']} | {s['total']} | {s['passed']} | " + f"{s.get('partial', 0)} | {s['failed']} | " f"{s['skipped']} | {s['errors']} | {s['discrepancies']} |") if s["discrepancies"] or s["errors"]: worst = max(worst, 1) diff --git a/tests/aws/redshift_fixtures.py b/tests/aws/redshift_fixtures.py new file mode 100644 index 0000000..66223e3 --- /dev/null +++ b/tests/aws/redshift_fixtures.py @@ -0,0 +1,239 @@ +"""Create, with Spark on EMR, the Iceberg tables Redshift cannot create itself. + +Redshift is a format-version 2 engine: CREATE TABLE ... ('format-version'='3') is +refused outright, and it has no DDL for branches, tags or equality deletes. On its +own that only tells us Redshift cannot *write* those things, which is not the same +question the matrix asks. A cell should distinguish + + "this engine cannot produce the feature" -> write gap + "this engine cannot even read the feature" -> no support at all + +so the features Redshift refuses to create are created here by Spark instead, and +the Redshift suite then tries to read and write them. A table it can read but not +create is partial support, not absent support. + +Runs as an EMR Serverless spark-submit entry point, driven by +run_redshift_fixtures.py. It writes a manifest next to the tables describing what +it actually managed to create, because that is release-dependent: the fixtures a +given Iceberg build refuses are exactly the ones the Redshift side must not claim +to have tested. + +Arguments (all passed as "--name value" by the driver): + --namespace + --manifest-uri s3://bucket/key.json where to write the manifest + --mode s3buckets|s3tables +""" + +import argparse +import json +import sys +import traceback + + +# Each fixture is (name, format_version, what it demonstrates, builder). +# A builder gets (spark, fqn) and raises if the runtime cannot express it; the +# failure is recorded in the manifest rather than aborting the job, so one +# unsupported fixture does not cost us the others. + +def _v3_basic(spark, fqn): + """A plain v3 table. The single most important fixture. + + If Redshift can read this, every v3 cell becomes a write gap rather than a + total absence, which is a different matrix answer. + """ + spark.sql(f"""CREATE TABLE {fqn} (id BIGINT, name STRING) + USING iceberg TBLPROPERTIES ('format-version'='3')""") + spark.sql(f"INSERT INTO {fqn} VALUES (1,'alpha'),(2,'beta'),(3,'gamma')") + + +def _v3_deletion_vectors(spark, fqn): + """A v3 table carrying deletion vectors. + + v3 replaces position-delete files with deletion vectors, so a DELETE on a v3 + merge-on-read table is enough to produce them. + """ + spark.sql(f"""CREATE TABLE {fqn} (id BIGINT, name STRING) + USING iceberg TBLPROPERTIES ( + 'format-version'='3', + 'write.delete.mode'='merge-on-read')""") + spark.sql(f"INSERT INTO {fqn} VALUES (1,'a'),(2,'b'),(3,'c'),(4,'d')") + spark.sql(f"DELETE FROM {fqn} WHERE id = 2") + + +def _v3_variant(spark, fqn): + """A v3 table with a VARIANT column.""" + spark.sql(f"""CREATE TABLE {fqn} (id BIGINT, payload VARIANT) + USING iceberg TBLPROPERTIES ('format-version'='3')""") + spark.sql(f"""INSERT INTO {fqn} + SELECT 1, parse_json('{{"a":1,"b":"two"}}')""") + + +def _v3_geometry(spark, fqn): + """A v3 table with a GEOMETRY column.""" + spark.sql(f"""CREATE TABLE {fqn} (id BIGINT, shape GEOMETRY) + USING iceberg TBLPROPERTIES ('format-version'='3')""") + spark.sql(f"INSERT INTO {fqn} SELECT 1, ST_Point(1.0, 2.0)") + + +def _v3_nanosecond(spark, fqn): + """A v3 table with a nanosecond-precision timestamp.""" + spark.sql(f"""CREATE TABLE {fqn} (id BIGINT, ts TIMESTAMP_NS) + USING iceberg TBLPROPERTIES ('format-version'='3')""") + spark.sql(f"INSERT INTO {fqn} VALUES (1, TIMESTAMP_NS '2026-01-01 00:00:00.123456789')") + + +def _v2_branch_tag(spark, fqn): + """A v2 table with a branch and a tag. + + Redshift has no branch or tag DDL. What matters for the cell is whether it can + still read the table, and whether it can address a branch at all. + """ + spark.sql(f"CREATE TABLE {fqn} (id BIGINT, name STRING) USING iceberg") + spark.sql(f"INSERT INTO {fqn} VALUES (1,'main-one'),(2,'main-two')") + spark.sql(f"ALTER TABLE {fqn} CREATE BRANCH audit_branch") + spark.sql(f"ALTER TABLE {fqn} CREATE TAG audit_tag") + # Diverge the branch so reading it gives a different answer from main, which + # is the only way to prove a branch read really happened. + spark.sql(f"INSERT INTO {fqn}.branch_audit_branch VALUES (3,'branch-only')") + + +def _v2_equality_deletes(spark, fqn): + """A v2 table carrying equality deletes. + + Spark writes position deletes, not equality deletes, so this is expected to + be unavailable here; it is attempted anyway rather than assumed, and the + manifest records the outcome so the Redshift side reports "not tested" + instead of inventing a result. + """ + spark.sql(f"""CREATE TABLE {fqn} (id BIGINT, name STRING) + USING iceberg TBLPROPERTIES ( + 'format-version'='2', + 'write.delete.mode'='merge-on-read', + 'write.update.mode'='merge-on-read', + 'write.merge.mode'='merge-on-read', + 'write.delete.granularity'='file')""") + spark.sql(f"INSERT INTO {fqn} VALUES (1,'a'),(2,'b'),(3,'c')") + spark.sql(f"DELETE FROM {fqn} WHERE id = 2") + # Confirm a delete file exists at all; whether it is an equality delete is + # read off the metadata by the checker below. + spark.sql(f"SELECT * FROM {fqn}.delete_files").collect() + + +FIXTURES = [ + ("fx_v3_basic", 3, "plain v3 table", _v3_basic), + ("fx_v3_dv", 3, "v3 deletion vectors", _v3_deletion_vectors), + ("fx_v3_variant", 3, "v3 VARIANT column", _v3_variant), + ("fx_v3_geometry", 3, "v3 GEOMETRY column", _v3_geometry), + ("fx_v3_ts_ns", 3, "v3 nanosecond timestamp", _v3_nanosecond), + ("fx_v2_branch", 2, "v2 branch and tag", _v2_branch_tag), + ("fx_v2_eqdel", 2, "v2 equality deletes", _v2_equality_deletes), +] + + +def describe(spark, fqn: str) -> dict: + """Read back what Iceberg actually stored, rather than what was asked for. + + A CREATE that succeeds does not prove the property took effect -- Redshift + itself silently discards PARTITIONED BY on S3 Tables -- so the fixture's real + format version and delete-file content are read from the metadata and shipped + in the manifest. That way the Redshift side can say "read a v3 table" only + when the table on disk is genuinely v3. + """ + facts = {} + try: + rows = spark.sql(f"SELECT * FROM {fqn}.metadata_log_entries").collect() + facts["snapshots"] = len(rows) + except Exception: # noqa: BLE001 + pass + try: + props = {r["key"]: r["value"] + for r in spark.sql(f"SHOW TBLPROPERTIES {fqn}").collect()} + facts["format_version"] = props.get("format-version", "?") + except Exception: # noqa: BLE001 + facts["format_version"] = "?" + try: + files = spark.sql( + f"SELECT content, count(*) AS n FROM {fqn}.delete_files GROUP BY content" + ).collect() + # Iceberg content codes: 1 = position deletes, 2 = equality deletes. + facts["delete_file_content"] = {str(r["content"]): r["n"] for r in files} + except Exception: # noqa: BLE001 + facts["delete_file_content"] = {} + try: + refs = spark.sql(f"SELECT name, type FROM {fqn}.refs").collect() + facts["refs"] = {r["name"]: r["type"] for r in refs} + except Exception: # noqa: BLE001 + facts["refs"] = {} + try: + facts["row_count"] = spark.sql(f"SELECT count(*) AS n FROM {fqn}").collect()[0]["n"] + except Exception: # noqa: BLE001 + pass + return facts + + +def main() -> int: + p = argparse.ArgumentParser() + p.add_argument("--namespace", required=True) + p.add_argument("--manifest-uri", required=True) + p.add_argument("--mode", required=True, choices=["s3buckets", "s3tables"]) + args, extra = p.parse_known_args() + if extra: + print(f"[fixtures] ignoring: {extra}") + + from pyspark.sql import SparkSession + + spark = SparkSession.builder.appName("redshift-fixtures").getOrCreate() + + print(f"[fixtures] creating namespace {args.namespace}") + spark.sql(f"CREATE NAMESPACE IF NOT EXISTS {args.namespace}") + + try: + version = spark.sql("SELECT 1").sparkSession.version + except Exception: # noqa: BLE001 + version = "?" + + manifest = { + "mode": args.mode, + "namespace": args.namespace, + "spark_version": version, + "fixtures": {}, + } + + for name, fmt, what, build in FIXTURES: + fqn = f"{args.namespace}.{name}" + entry = {"format_version_requested": fmt, "describes": what} + try: + spark.sql(f"DROP TABLE IF EXISTS {fqn} PURGE") + except Exception: # noqa: BLE001 + pass + try: + build(spark, fqn) + entry["created"] = True + entry["stored"] = describe(spark, fqn) + print(f"[fixtures] OK {name}: {entry['stored']}") + except Exception as e: # noqa: BLE001 - a refused fixture is a result + entry["created"] = False + entry["error"] = f"{type(e).__name__}: {e}"[:400] + print(f"[fixtures] FAIL {name}: {entry['error']}") + print(traceback.format_exc()[:1500]) + manifest["fixtures"][name] = entry + + created = [n for n, e in manifest["fixtures"].items() if e.get("created")] + print(f"[fixtures] created {len(created)}/{len(FIXTURES)}: {', '.join(created)}") + + import boto3 + from urllib.parse import urlparse + + parsed = urlparse(args.manifest_uri) + boto3.client("s3").put_object( + Bucket=parsed.netloc, + Key=parsed.path.lstrip("/"), + Body=json.dumps(manifest, indent=2).encode(), + ContentType="application/json", + ) + print(f"[fixtures] manifest written to {args.manifest_uri}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/aws/run_redshift.py b/tests/aws/run_redshift.py new file mode 100644 index 0000000..92cec69 --- /dev/null +++ b/tests/aws/run_redshift.py @@ -0,0 +1,199 @@ +"""Run the Redshift Iceberg feature suite against Redshift Serverless, both modes. + +Unlike the EMR and Glue drivers, this one does not ship code to a cluster. Redshift +is driven entirely through the Data API, so the suite runs here in the runner and +every statement is a redshift-data call. That removes the bundle, the entry point +and the report round-trip through S3, which is why this driver is much shorter +despite covering the same 70 checks. + +What it still shares with the other drivers is the output: the reports are renamed +into the --iceberg-test-report.* convention and handed to +platform_common.summarise(), so a Redshift run publishes the same job summary +shape as every other engine. That is the whole point of driving managed engines +this way and it must not drift. + +Environment: + AWS_REGION (required) + REDSHIFT_WORKGROUP (default: -wg) + REDSHIFT_DATABASE (default: dev) + REDSHIFT_NAMESPACE (default: -ns) used to look the secret up + REDSHIFT_SECRET_ARN admin secret; looked up from the namespace if unset + REDSHIFT_ROLE_ARN (required) role the external schema names + AWS_DATA_BUCKET (required for the s3buckets mode) + AWS_TABLE_BUCKET_ARN (required for the s3tables mode) + REDSHIFT_FIXTURE_DB optional Spark-created fixtures (see redshift_fixtures.py) + RUN_TAG (required) unique per run + MODES both|s3buckets|s3tables (default: both) +""" + +import json +import os +import runpy +import shutil +import sys +from pathlib import Path + +import boto3 + +from platform_common import ( # noqa: E402 - sibling module, not a package + DATA_BUCKET, + LOCAL_REPORT_DIR, + REGION, + RESOURCE_PREFIX, + REPO_ROOT, + RUN_TAG, + TABLE_BUCKET_ARN, + modes_to_run, + summarise, +) + +ENGINE = "redshift" +WORKGROUP = os.environ.get("REDSHIFT_WORKGROUP", f"{RESOURCE_PREFIX}-wg") +NAMESPACE = os.environ.get("REDSHIFT_NAMESPACE", f"{RESOURCE_PREFIX}-ns") +DATABASE = os.environ.get("REDSHIFT_DATABASE", "dev") +ROLE_ARN = os.environ.get("REDSHIFT_ROLE_ARN", "") +FIXTURE_DB = os.environ.get("REDSHIFT_FIXTURE_DB", "") + +SUITE = REPO_ROOT / "tests" / "redshift_feature_tests.py" + + +def admin_secret_arn() -> str: + """The namespace's managed admin secret, looked up rather than configured. + + CloudFormation cannot return it: AWS::RedshiftServerless::Namespace rejects + !GetAtt Namespace.AdminPasswordSecretArn as "must be a readonly property in + schema", so the stack has no output to wire into a GitHub secret. The API does + expose it, and with ManageAdminPassword the ARN is stable for the namespace, so + resolving it here keeps the credential out of the template, the workflow inputs + and the repository. + """ + configured = os.environ.get("REDSHIFT_SECRET_ARN", "") + if configured: + return configured + ns = boto3.client("redshift-serverless", region_name=REGION).get_namespace( + namespaceName=NAMESPACE + )["namespace"] + arn = ns.get("adminPasswordSecretArn", "") + if not arn: + raise SystemExit( + f"namespace {NAMESPACE} has no adminPasswordSecretArn; either it was " + "not created with ManageAdminPassword or REDSHIFT_SECRET_ARN must be set" + ) + print(f"[driver] resolved admin secret for namespace {NAMESPACE}") + return arn + + +def engine_version() -> str: + """Redshift's own version string, for the report. + + Best-effort: the suite detects it too, and a driver that died here would be + failing the run over a cosmetic field. + """ + try: + wg = boto3.client("redshift-serverless", region_name=REGION).get_workgroup( + workgroupName=WORKGROUP + )["workgroup"] + return f"{wg.get('baseCapacity', '?')} RPU" + except Exception as e: # noqa: BLE001 + print(f"[driver] could not describe workgroup: {type(e).__name__}: {e}") + return "unknown" + + +def run_mode(mode: str, secret_arn: str) -> dict: + """Run the suite in-process for one storage mode. + + runpy rather than a subprocess so a crash surfaces here with a real traceback + instead of an exit code, and so the report paths need no plumbing. The suite + calls sys.exit() to signal discrepancies, which is a result and not a driver + failure, so SystemExit is caught. + """ + data_path = f"src/data/platforms/aws/{mode}/redshift-s3/redshift-s3.json" + print(f"\n[driver] === {mode}: workgroup {WORKGROUP}, database {DATABASE} ===") + + os.environ.update({ + "REPO_ROOT": str(REPO_ROOT), + "REPORT_DIR": str(LOCAL_REPORT_DIR), + "AWS_REGION": REGION, + "REDSHIFT_WORKGROUP": WORKGROUP, + "REDSHIFT_DATABASE": DATABASE, + "REDSHIFT_SECRET_ARN": secret_arn, + "REDSHIFT_ROLE_ARN": ROLE_ARN, + "AWS_DATA_BUCKET": DATA_BUCKET, + "AWS_TABLE_BUCKET_ARN": TABLE_BUCKET_ARN, + "MATRIX_STORAGE_MODE": mode, + "MATRIX_DATA_PATH": data_path, + "MATRIX_NS_PREFIX": RESOURCE_PREFIX, + "PLATFORM_LABEL": f"Redshift Serverless {engine_version()} / {mode}", + "REDSHIFT_FIXTURE_DB": FIXTURE_DB, + # One tag per mode. The two runs own separate external schemas, Glue + # databases and S3 Tables namespaces, so a shared tag would have the + # second mode's teardown delete the first mode's objects. Hyphens become + # underscores because the tag ends up inside SQL identifiers. + "RUN_TAG": f"{RUN_TAG}_{mode}".replace("-", "_").lower(), + }) + + exit_code = 0 + try: + runpy.run_path(str(SUITE), run_name="__main__") + except SystemExit as e: + exit_code = int(e.code or 0) + + # The suite names its own artefacts; rename them into the shared convention so + # summarise() and the other engines' reports stay interchangeable. + produced = LOCAL_REPORT_DIR / f"redshift-iceberg-test-report-{mode}.json" + report = None + if produced.is_file(): + for suffix in (".json", ".md"): + src = produced.with_suffix(suffix) + if src.is_file(): + shutil.move(str(src), + str(LOCAL_REPORT_DIR / f"{ENGINE}-{mode}-iceberg-test-report{suffix}")) + report = json.loads( + (LOCAL_REPORT_DIR / f"{ENGINE}-{mode}-iceberg-test-report.json").read_text() + ) + print(f"[driver] {mode}: {json.dumps(report['summary'])}") + else: + print(f"[driver] {mode} produced no report at {produced}") + + return {"mode": mode, "job_run_id": "", "report": report, + "state": "SUCCESS" if report else "NO_REPORT", + "state_details": f"suite exit {exit_code}", "report_uri": ""} + + +def main() -> int: + if not ROLE_ARN: + raise SystemExit("REDSHIFT_ROLE_ARN is required: the external schema must " + "name a role, because the auto-mounted catalog cannot write") + modes = modes_to_run() + print(f"[driver] region={REGION} workgroup={WORKGROUP} modes={modes}") + print(f"[driver] fixtures: {FIXTURE_DB or 'none configured'}") + + secret_arn = admin_secret_arn() + LOCAL_REPORT_DIR.mkdir(parents=True, exist_ok=True) + + results, reports = [], {} + for mode in modes: + try: + r = run_mode(mode, secret_arn) + except Exception as e: # noqa: BLE001 - one mode must not hide the other + print(f"[driver] {mode} raised: {type(e).__name__}: {e}") + results.append({"mode": mode, "job_run_id": "", "state": "DRIVER_ERROR", + "state_details": str(e)[:300], "report_uri": ""}) + continue + results.append(r) + if r["report"]: + reports[mode] = r["report"] + + return summarise( + ENGINE, + "AWS Redshift Serverless Iceberg Feature Test Report", + [f"- **Workgroup:** {WORKGROUP} ({engine_version()})", + f"- **Catalog:** Glue Data Catalog external schema with IAM_ROLE", + f"- **Fixtures:** {FIXTURE_DB or 'none (features Redshift cannot create report as unmeasured)'}", + f"- **Run:** {RUN_TAG}"], + results, reports, + ) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/aws/run_redshift_fixtures.py b/tests/aws/run_redshift_fixtures.py new file mode 100644 index 0000000..1f7a4db --- /dev/null +++ b/tests/aws/run_redshift_fixtures.py @@ -0,0 +1,158 @@ +"""Submit the Redshift fixture job to EMR Serverless and report the manifest. + +Creates, with Spark, the Iceberg tables Redshift refuses to create, so the +Redshift suite can tell "cannot create" from "cannot read". See +redshift_fixtures.py for what is built and why. + +Reuses the EMR plumbing in platform_common.py and run_emr_serverless.py rather +than duplicating it: the same application, the same S3 layout, the same catalog +wiring for both storage modes. The only difference is the entry point and that +this job writes a manifest instead of a report. + +Environment: + AWS_REGION (required) + AWS_DATA_BUCKET (required) + AWS_EMR_JOB_ROLE_ARN (required) + AWS_TABLE_BUCKET_ARN (required for the s3tables mode) + RUN_TAG (required) + MODES both|s3buckets|s3tables (default: both) + EMR_APPLICATION_ID reuse an existing application instead of creating one + EMR_RELEASE_LABEL (default: emr-spark-8.0.0) +""" + +import json +import sys +from pathlib import Path + +import boto3 + +from platform_common import ( # noqa: E402 - sibling module, not a package + DATA_BUCKET, + GLUE_CATALOG_IMPL, + JOB_TIMEOUT_MINUTES, + REGION, + RESOURCE_PREFIX, + RUN_TAG, + TABLE_BUCKET_ARN, + modes_to_run, + s3_uri, + s3tables_catalog_props, + upload, + wait_for, +) +from run_emr_serverless import ( # noqa: E402 + JOB_ROLE_ARN, + RELEASE_LABEL, + create_application, + spark_submit_params, +) + +ENGINE = "emr" +emr = boto3.client("emr-serverless", region_name=REGION) +s3 = boto3.client("s3", region_name=REGION) + +# The namespace the fixtures live in. Deliberately stable rather than per-run: +# the Redshift suite has to find it, and a fixture set is cheap to leave in place +# between the two runs of a single CI job. tests/aws/teardown.py removes it. +FIXTURE_NAMESPACE = f"{RESOURCE_PREFIX}_rsfix" + + +def manifest_key(mode: str) -> str: + return f"{ENGINE}/fixtures/{RUN_TAG}/{mode}/manifest.json" + + +def run_mode(app_id: str, mode: str, entry_uri: str) -> dict: + warehouse = s3_uri(ENGINE, "fixtures-warehouse", RUN_TAG) + "/" + if mode == "s3tables": + if not TABLE_BUCKET_ARN: + raise SystemExit("AWS_TABLE_BUCKET_ARN is required for the s3tables mode") + catalog_props = s3tables_catalog_props() + else: + catalog_props = "" + + manifest_uri = f"s3://{DATA_BUCKET}/{manifest_key(mode)}" + print(f"\n[fixtures] === {mode}: namespace {FIXTURE_NAMESPACE} ===") + + resp = emr.start_job_run( + applicationId=app_id, + executionRoleArn=JOB_ROLE_ARN, + name=f"{RESOURCE_PREFIX}-rsfix-{mode}"[:64], + executionTimeoutMinutes=JOB_TIMEOUT_MINUTES, + jobDriver={ + "sparkSubmit": { + "entryPoint": entry_uri, + "entryPointArguments": [ + "--namespace", FIXTURE_NAMESPACE, + "--manifest-uri", manifest_uri, + "--mode", mode, + ], + "sparkSubmitParameters": spark_submit_params( + mode, GLUE_CATALOG_IMPL, warehouse, catalog_props, ""), + } + }, + configurationOverrides={ + "monitoringConfiguration": { + "s3MonitoringConfiguration": { + "logUri": s3_uri(ENGINE, "logs", RUN_TAG) + "/" + } + } + }, + tags={"project": "iceberg-matrix", "run": RUN_TAG, "mode": f"rsfix-{mode}"}, + ) + job_id = resp["jobRunId"] + state = wait_for( + lambda: emr.get_job_run(applicationId=app_id, jobRunId=job_id)["jobRun"]["state"], + want={"SUCCESS", "FAILED", "CANCELLED"}, bad=set(), + what=f"fixture job {job_id} ({mode})", + timeout_s=JOB_TIMEOUT_MINUTES * 60 + 300, + ) + details = emr.get_job_run(applicationId=app_id, jobRunId=job_id)["jobRun"] + if state != "SUCCESS": + print(f"[fixtures] job {job_id} {state}: {details.get('stateDetails', '')}") + print(f"[fixtures] logs: {s3_uri(ENGINE, 'logs', RUN_TAG)}" + f"/applications/{app_id}/jobs/{job_id}/") + + manifest = None + try: + body = s3.get_object(Bucket=DATA_BUCKET, Key=manifest_key(mode))["Body"].read() + manifest = json.loads(body) + except Exception as e: # noqa: BLE001 + print(f"[fixtures] no manifest for {mode}: {type(e).__name__}: {e}") + + return {"mode": mode, "job_run_id": job_id, "state": state, + "manifest_uri": manifest_uri, "manifest": manifest} + + +def main() -> int: + modes = modes_to_run() + print(f"[fixtures] region={REGION} bucket={DATA_BUCKET} modes={modes}") + + entry_uri = upload(Path(__file__).with_name("redshift_fixtures.py"), + f"{ENGINE}/scripts/{RUN_TAG}/redshift_fixtures.py") + app_id = create_application() + Path("/tmp/emr-application-id").write_text(app_id) + + worst = 0 + for mode in modes: + result = run_mode(app_id, mode, entry_uri) + if result["state"] != "SUCCESS" or not result["manifest"]: + worst = 1 + continue + fixtures = result["manifest"]["fixtures"] + made = [n for n, e in fixtures.items() if e.get("created")] + missed = {n: e.get("error", "") for n, e in fixtures.items() + if not e.get("created")} + print(f"\n[fixtures] {mode}: created {len(made)}/{len(fixtures)}") + for name in made: + print(f" OK {name}: {fixtures[name].get('stored')}") + for name, err in missed.items(): + print(f" MISS {name}: {err[:200]}") + print(f"[fixtures] manifest: {result['manifest_uri']}") + print(f"[fixtures] point the suite at it with " + f"REDSHIFT_FIXTURE_MANIFEST={result['manifest_uri']}") + + return worst + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/aws/teardown.py b/tests/aws/teardown.py index 3026b72..a5ad86f 100644 --- a/tests/aws/teardown.py +++ b/tests/aws/teardown.py @@ -208,8 +208,21 @@ def delete_s3_prefixes() -> None: return s3 = boto3.client("s3", region_name=REGION) # reports/ is intentionally left behind; lifecycle expires it. - for kind in ("warehouse", "logs", "scripts"): - prefix = f"{ENGINE}/{kind}/{RUN_TAG}/" + # + # fixtures-warehouse holds real Parquet for the Spark-built Redshift fixtures + # and bills like any other warehouse, so it has to be swept even when the + # engine under test was not EMR. The Redshift suite cleans its own + # redshift// prefix on the way out, but a crashed run would not, so it is + # listed here too: deleting an absent prefix is free. + prefixes = [f"{ENGINE}/{kind}/{RUN_TAG}/" + for kind in ("warehouse", "logs", "scripts", + "fixtures", "fixtures-warehouse")] + # Deliberately not "/"-terminated. The Redshift driver derives one tag per + # storage mode by sanitising the run tag for use in SQL identifiers + # (icebergmatrix-123 -> icebergmatrix_123_s3buckets), so this has to match a + # tag prefix rather than an exact directory. + prefixes.append(f"redshift/{RUN_TAG.replace('-', '_').lower()}") + for prefix in prefixes: deleted = 0 try: pages = s3.get_paginator("list_objects_v2").paginate(Bucket=DATA_BUCKET, Prefix=prefix) diff --git a/tests/flink_feature_tests.py b/tests/flink_feature_tests.py index a55f537..f329357 100644 --- a/tests/flink_feature_tests.py +++ b/tests/flink_feature_tests.py @@ -100,6 +100,22 @@ def _default_host() -> str: VERSIONS = ["v2", "v3"] +# Which platform's matrix cells to compare against. Overridable so the same +# suite can be pointed at another platform that runs this engine, matching how +# the Spark-based suites work. +MATRIX_PLATFORM_ID = os.environ.get("MATRIX_PLATFORM_ID", "flink") +MATRIX_DATA_PATH = os.environ.get( + "MATRIX_DATA_PATH", "src/data/platforms/oss/flink/flink.json" +) +# Free-text label for the runtime under test, recorded in the report so a cell +# that changes later is attributable. +PLATFORM_LABEL = os.environ.get("PLATFORM_LABEL", "") +# How the catalog under test is reached, recorded in the report. Overridable so +# a managed platform can describe its own catalog instead of the local stack. +CATALOG_MODE = os.environ.get( + "MATRIX_CATALOG_MODE", f"REST ({REST_URI}, warehouse={REST_WAREHOUSE})" +) + # How long to wait for in-job compaction to produce a rewrite commit. Sized off # the 5s checkpoint interval configured in docker-compose.flink.yml: a rewrite # was observed within ~90s locally. @@ -1768,18 +1784,56 @@ def test_snowflake_horizon_catalog(version: str) -> TestResult: # --------------------------------------------------------------------------- def load_flink_json_support() -> dict: - """Load the recorded support levels for Flink from the matrix data.""" - path = os.path.join(REPO_ROOT, "src", "data", "platforms", "oss", "flink", "flink.json") + """Load the recorded support levels for the platform under test. + + Reads MATRIX_DATA_PATH and keeps only the cells belonging to + MATRIX_PLATFORM_ID, so the suite is not tied to the OSS Flink cells. + """ + path = os.path.join(REPO_ROOT, MATRIX_DATA_PATH) with open(path) as f: data = json.load(f) result = {} for key, val in data.get("support", {}).items(): parts = key.split(":") - if len(parts) == 3 and parts[0] == "flink": + if len(parts) == 3 and parts[0] == MATRIX_PLATFORM_ID: result[(parts[1], parts[2])] = val.get("level", "unknown") return result +def load_matrix_features() -> dict: + """Load feature definitions from the matrix source of truth (features.json). + + Lets the suite assert that EVERY feature shown in the matrix is exercised, so + a newly added matrix feature cannot silently go untested. + """ + with open(os.path.join(REPO_ROOT, "src", "data", "features.json")) as f: + data = json.load(f) + return { + feat["id"]: { + "name": feat.get("name", feat["id"]), + "introducedIn": feat.get("introducedIn", "v2"), + } + for feat in data.get("features", []) + } + + +def compute_coverage(results: list) -> dict: + """Compare the set of tested feature ids against the matrix features. + + A non-empty "uncovered" list means the suite has drifted from the matrix and + is treated as a failure, matching the Spark-based suites. + """ + matrix = load_matrix_features() + tested = {r.feature_id for r in results} + uncovered = sorted(set(matrix) - tested) + return { + "matrix_feature_count": len(matrix), + "tested_feature_count": len(tested), + "uncovered": [{"id": fid, "name": matrix[fid]["name"]} for fid in uncovered], + "extra": sorted(tested - set(matrix)), + } + + def compute_match(test_result: str, json_level: str) -> bool: """Whether an executed test agrees with the recorded support level. @@ -1818,12 +1872,19 @@ def generate_report(results: list) -> dict: "verified": not is_unverified, }) + coverage = compute_coverage(results) + return { "timestamp": datetime.now(tz=timezone.utc).isoformat(), "engine": "Flink", "mode": MODE, "flink_version": FLINK_VERSION, "flink_iceberg_version": FLINK_ICEBERG_VERSION, + "platform": MATRIX_PLATFORM_ID, + "platform_label": PLATFORM_LABEL, + "catalog_mode": CATALOG_MODE, + "versions_tested": VERSIONS, + "coverage": coverage, "tests": tests_output, "summary": { "total": len(results), @@ -1833,6 +1894,7 @@ def generate_report(results: list) -> dict: "errors": sum(1 for r in results if r.result == "error"), "discrepancies": discrepancies, "unverified": unverified, + "uncovered_features": len(coverage["uncovered"]), }, } @@ -1846,6 +1908,12 @@ def generate_markdown(report: dict) -> str: f"- **Flink Version:** {report['flink_version']}", f"- **Iceberg Version:** {report['flink_iceberg_version']}", f"- **Execution mode:** {report['mode']}", + f"- **Catalog:** {report.get('catalog_mode', 'unknown')}", + ] + if report.get("platform_label"): + lines.append(f"- **Platform:** {report['platform_label']}") + lines += [ + f"- **Format Versions Tested:** {', '.join(report.get('versions_tested', []))}", "", "## Summary", "", @@ -1858,11 +1926,33 @@ def generate_markdown(report: dict) -> str: f"| Errors | {s['errors']} |", f"| Discrepancies vs matrix | {s['discrepancies']} |", f"| Unverified (skip/error) | {s['unverified']} |", + f"| Uncovered matrix features | {s.get('uncovered_features', 0)} |", "", "`Failed` is a result, not a defect: it records that the engine does not " "support the feature through Flink SQL. A discrepancy means the observed " "behaviour disagrees with `flink.json`.", "", + ] + + cov = report.get("coverage") + if cov: + lines.append( + f"**Matrix coverage:** {cov['tested_feature_count']}/" + f"{cov['matrix_feature_count']} features in `features.json` have a test." + ) + if cov["uncovered"]: + lines += ["", "### Uncovered matrix features (no test!)", ""] + for f in cov["uncovered"]: + lines.append( + f"- **{f['name']}** (`{f['id']}`) - add a `test_*` function " + "and register it in `ALL_TESTS`" + ) + if cov.get("extra"): + lines += ["", f"> Note: tests exist for ids not in the matrix: " + f"{', '.join(cov['extra'])}"] + lines.append("") + + lines += [ "## Test Results", "", "| Feature | Version | Result | Matrix | Match | Details |", @@ -1915,6 +2005,10 @@ def main(): print(f"REST catalog: {REST_URI} (warehouse {REST_WAREHOUSE})") print(f"S3 endpoint: {S3_ENDPOINT}") print(f"JDBC catalog: {JDBC_URI}") + print(f"Matrix platform: {MATRIX_PLATFORM_ID} ({MATRIX_DATA_PATH})") + if PLATFORM_LABEL: + print(f"Platform: {PLATFORM_LABEL}") + print(f"Versions: {', '.join(VERSIONS)}") print() if MODE == "local" and not FLINK_HOME: @@ -1963,7 +2057,8 @@ def main(): print(f"\n{'=' * 70}") print(f" {s['passed']} passed, {s['failed']} failed, {s['skipped']} skipped, " f"{s['errors']} errors, {s['discrepancies']} discrepancies, " - f"{s['unverified']} unverified") + f"{s['unverified']} unverified, " + f"{s.get('uncovered_features', 0)} uncovered matrix features") print(f" Reports: {json_path}") print(f" {md_path}") print(f"{'=' * 70}") @@ -1975,8 +2070,10 @@ def main(): f.write(md_content) # Errors mean the harness itself could not run something; discrepancies mean - # the matrix and reality disagree. Both warrant a human look. - sys.exit(1 if (s["discrepancies"] > 0 or s["errors"] > 0) else 0) + # the matrix and reality disagree; uncovered features mean the suite has + # drifted from the matrix. All three warrant a human look. + sys.exit(1 if (s["discrepancies"] > 0 or s["errors"] > 0 + or s.get("uncovered_features", 0) > 0) else 0) if __name__ == "__main__": diff --git a/tests/redshift_feature_tests.py b/tests/redshift_feature_tests.py new file mode 100644 index 0000000..16ff3e9 --- /dev/null +++ b/tests/redshift_feature_tests.py @@ -0,0 +1,2243 @@ +#!/usr/bin/env python3 +""" +Redshift-based Iceberg Feature Test Suite (V2 + V3). + +Drives Amazon Redshift Serverless through the Redshift Data API and compares what +the engine actually does with the support levels recorded for the +aws-redshift-s3 platform. Disagreements are reported as "discrepancies"; +features that genuinely cannot be exercised from Redshift SQL are reported as +"skip" with an honest reason and counted as "unverified", so a skip can never +silently rubber-stamp the matrix. + +Why the Data API rather than a database connection: statements are submitted as +AWS API calls, so there is no JDBC driver, no password and no network path to +arrange. Authentication uses the Redshift-managed admin secret, which gives a +deterministic identity -- the alternative, the caller's own IAM identity, lands +as a database user whose grants differ between a laptop and CI. + +Two things about Redshift shape this suite, both measured rather than assumed: + + Writes need an external schema that names an IAM role. Creating an Iceberg + table through the auto-mounted awsdatacatalog fails with "No session + credential found": that path authorises data access with the caller's IAM + session, and a Data API connection authenticated as a database user has + none. This is the documented "federated identity is not supported when + writing to Apache Iceberg tables" limitation. + + Redshift is an Iceberg v2 engine. CREATE TABLE with + 'format-version'='3' is rejected outright, so every V3 feature is a + measured failure rather than an untested guess. + +Features Redshift cannot create at all are handled the way the Flink suite +handles them: an EMR-created fixture table is read (and written) instead, which +separates "cannot create" from "cannot read" and justifies partial support. + +Usage: + export REDSHIFT_WORKGROUP=icebergmatrix-wg + export REDSHIFT_SECRET_ARN=arn:aws:secretsmanager:...:secret:redshift!... + export REDSHIFT_ROLE_ARN=arn:aws:iam::...:role/icebergmatrix-redshift + export AWS_DATA_BUCKET=iceberg-tests-matrix + python tests/redshift_feature_tests.py + +Environment variables: + REDSHIFT_WORKGROUP Serverless workgroup name + REDSHIFT_DATABASE database inside the namespace (default: dev) + REDSHIFT_SECRET_ARN Redshift-managed admin secret; omit to use the + caller's IAM identity instead + REDSHIFT_ROLE_ARN IAM role named on the external schema + AWS_DATA_BUCKET bucket holding table data for the s3buckets mode + AWS_TABLE_BUCKET_ARN S3 Tables bucket ARN for the s3tables mode + MATRIX_STORAGE_MODE s3buckets | s3tables (default: s3buckets) + MATRIX_PLATFORM_ID platform whose cells to compare against + MATRIX_DATA_PATH matrix file holding those cells + MATRIX_NS_PREFIX prefix for created Glue databases and schemas + REDSHIFT_FIXTURE_DB Glue database holding EMR-created fixture tables + PLATFORM_LABEL free-text label recorded in the report + REPO_ROOT / REPORT_DIR repo root and where reports are written + REDSHIFT_ONLY comma-separated test names, for iterating on one +""" + +import json +import os +import re +import sys +import time +import uuid +from datetime import datetime, timezone +from pathlib import Path + +# --------------------------------------------------------------------------- +# Configuration +# --------------------------------------------------------------------------- +REPO_ROOT = os.environ.get("REPO_ROOT", str(Path(__file__).resolve().parent.parent)) +REPORT_DIR = os.environ.get("REPORT_DIR", os.path.join(os.getcwd(), "test-reports")) + +REGION = os.environ.get("AWS_REGION", "us-east-1") +WORKGROUP = os.environ.get("REDSHIFT_WORKGROUP", "icebergmatrix-wg") +DATABASE = os.environ.get("REDSHIFT_DATABASE", "dev") +SECRET_ARN = os.environ.get("REDSHIFT_SECRET_ARN", "") +ROLE_ARN = os.environ.get("REDSHIFT_ROLE_ARN", "") +DATA_BUCKET = os.environ.get("AWS_DATA_BUCKET", "") +TABLE_BUCKET_ARN = os.environ.get("AWS_TABLE_BUCKET_ARN", "") + +STORAGE_MODE = os.environ.get("MATRIX_STORAGE_MODE", "s3buckets") +NS_PREFIX = os.environ.get("MATRIX_NS_PREFIX", "icebergmatrix") +RUN_TAG = os.environ.get("RUN_TAG", uuid.uuid4().hex[:8]) + +# Catalog objects this run owns. All are dropped on the way out. +GLUE_DB = f"{NS_PREFIX}_rs_{RUN_TAG}" +SCHEMA = f"rs_{RUN_TAG}" +# s3tables mode only: the namespace inside the table bucket, and the resource +# link in the default Glue catalog that points at it. Redshift cannot name a +# federated catalog directly, so the link is what makes the namespace reachable. +S3T_NAMESPACE = f"{NS_PREFIX}_rs_{RUN_TAG}" +RESOURCE_LINK = f"{NS_PREFIX}_rslink_{RUN_TAG}" +# Tables Spark created on EMR for the features Redshift cannot create itself. +# Without these, "Redshift cannot do X" conflates two different answers: cannot +# write X, and cannot even read X. See tests/aws/redshift_fixtures.py. +FIXTURE_DB = os.environ.get("REDSHIFT_FIXTURE_DB", "") +FIXTURE_SCHEMA = f"rsfix_{RUN_TAG}" +FIXTURE_LINK = f"{NS_PREFIX}_rsfixlink_{RUN_TAG}" +# Fixture table names, kept in step with tests/aws/redshift_fixtures.py. +FX_V3_BASIC = "fx_v3_basic" +FX_V3_DV = "fx_v3_dv" +FX_V3_VARIANT = "fx_v3_variant" +FX_V3_GEOMETRY = "fx_v3_geometry" +FX_V3_TS_NS = "fx_v3_ts_ns" +FX_V2_BRANCH = "fx_v2_branch" +FX_V2_EQDEL = "fx_v2_eqdel" + +MATRIX_PLATFORM_ID = os.environ.get("MATRIX_PLATFORM_ID", "aws-redshift-s3") +MATRIX_DATA_PATH = os.environ.get( + "MATRIX_DATA_PATH", + f"src/data/platforms/aws/{STORAGE_MODE}/redshift-s3/redshift-s3.json", +) +PLATFORM_LABEL = os.environ.get("PLATFORM_LABEL", "") +REDSHIFT_VERSION = os.environ.get("REDSHIFT_VERSION", "unknown") + +VERSIONS = ["v2", "v3"] +STATEMENT_TIMEOUT = int(os.environ.get("REDSHIFT_STATEMENT_TIMEOUT", "300")) +# Write probes against a fixture only need long enough to be refused. A DELETE +# against a v3 deletion-vector table was observed to hang until the full statement +# timeout instead of erroring, so these are capped much lower. +FIXTURE_WRITE_TIMEOUT = int(os.environ.get("REDSHIFT_FIXTURE_WRITE_TIMEOUT", "45")) + +CATALOG_MODE = os.environ.get( + "MATRIX_CATALOG_MODE", + f"Glue Data Catalog external schema (IAM_ROLE), mode={STORAGE_MODE}", +) + + +# --------------------------------------------------------------------------- +# Data API plumbing +# --------------------------------------------------------------------------- + +def _client(service): + import boto3 + return boto3.client(service, region_name=REGION) + + +_DATA = None + + +def _data(): + global _DATA + if _DATA is None: + _DATA = _client("redshift-data") + return _DATA + + +def _run_sql(statements, timeout: int = None) -> tuple: + """Run statements in order, stopping at the first failure. + + Returns (ok, output) where output concatenates any result rows, so the tests + can assert on content the same way the other engine suites do. + """ + timeout = timeout or STATEMENT_TIMEOUT + chunks = [] + for sql in statements: + sql = sql.strip().rstrip(";") + if not sql: + continue + ok, detail, rows = _execute(sql, timeout) + if not ok: + # Deliberately not echoing the statement: several tests substring + # match the output, and the SQL text would create false matches. + chunks.append(detail) + return False, "\n".join(chunks).strip() + for row in rows: + chunks.append(" | ".join("" if v is None else str(v) for v in row)) + return True, "\n".join(chunks).strip() + + +def _execute(sql: str, timeout: int) -> tuple: + """Submit one statement and wait for a terminal state.""" + kwargs = {"WorkgroupName": WORKGROUP, "Database": DATABASE, "Sql": sql} + if SECRET_ARN: + kwargs["SecretArn"] = SECRET_ARN + try: + sid = _data().execute_statement(**kwargs)["Id"] + except Exception as e: # noqa: BLE001 - any submit failure is a result + return False, f"{type(e).__name__}: {e}", [] + + deadline = time.time() + timeout + while time.time() < deadline: + try: + described = _data().describe_statement(Id=sid) + except Exception as e: # noqa: BLE001 + return False, f"describe failed: {type(e).__name__}: {e}", [] + status = described["Status"] + if status == "FINISHED": + if not described.get("HasResultSet"): + return True, "", [] + return True, "", _fetch_rows(sid) + if status in ("FAILED", "ABORTED"): + return False, str(described.get("Error", status)), [] + time.sleep(1.5) + return False, f"statement timed out after {timeout}s", [] + + +def _fetch_rows(sid: str) -> list: + """Flatten a Data API result set into plain Python rows.""" + rows = [] + try: + page = _data().get_statement_result(Id=sid) + except Exception: # noqa: BLE001 - a result we cannot read is not a failure + return rows + while True: + for record in page.get("Records", []): + row = [] + for cell in record: + if not cell or cell.get("isNull"): + row.append(None) + else: + row.append(list(cell.values())[0]) + rows.append(row) + token = page.get("NextToken") + if not token: + break + page = _data().get_statement_result(Id=sid, NextToken=token) + return rows + + +def _unique(prefix: str = "t") -> str: + return f"{prefix}_{uuid.uuid4().hex[:8]}" + + +def _fmt(version: str) -> str: + """Iceberg format-version number for a matrix version label.""" + return "3" if version == "v3" else "2" + + +def _loc(name: str) -> str: + """LOCATION clause for a table in the s3buckets mode. + + S3 Tables determines its own location, so the clause is omitted there. + """ + if STORAGE_MODE == "s3tables": + return "" + return f"LOCATION 's3://{DATA_BUCKET}/redshift/{RUN_TAG}/{name}/'" + + +def _table(name: str) -> str: + return f"{SCHEMA}.{name}" + + +def _error_reason(out: str, limit: int = 220) -> str: + """Condense an engine error to its most informative part. + + Redshift reports some failures on one line, and others as a block whose first + line is a bare "ERROR:" followed by a rule and then the substance: + + ERROR: + ----------------------------------------------- + error: Error parsing table metadata. + code: 15003 + context: Invalid column type. Got: variant + + Returning the first line containing "ERROR:" throws the substance away and + records the useless string "ERROR:" as the evidence for a cell, so the + informative "error:" and "context:" fields are pulled out and joined instead. + """ + if not out: + return "no output" + lines = [l.strip() for l in out.splitlines() if l.strip()] + interesting = [ + l for l in lines + if l.lower().startswith(("error:", "context:", "detail:")) + and l.strip().lower() not in ("error:",) + ] + if interesting: + # Deduplicate while keeping order: "error:" and "context:" often overlap. + seen, parts = set(), [] + for line in interesting: + if line not in seen: + seen.add(line) + parts.append(line) + return " | ".join(parts)[:limit] + for line in lines: + if ("ERROR:" in line or "Exception" in line) and line.strip() != "ERROR:": + return line[:limit] + return lines[0][:limit] if lines else out[:limit] + + +# --------------------------------------------------------------------------- +# Result class +# --------------------------------------------------------------------------- + +class TestResult: + def __init__(self, feature_id: str, feature_name: str, version: str = "v2"): + self.feature_id = feature_id + self.feature_name = feature_name + # partial means measured as genuinely half-supported, e.g. readable but + # not writable. It is a positive finding, not a shorthand for "unsure": + # anything unmeasured is a skip. + self.result = "skip" # pass | partial | fail | skip | error + self.details = "" + self.version_tested = version + + def to_dict(self): + return { + "feature_id": self.feature_id, + "feature_name": self.feature_name, + "version": self.version_tested, + "result": self.result, + "details": self.details, + } + + +def _v3_unsupported(feature_id: str, feature_name: str, version: str) -> TestResult: + """A V3 feature on an engine that cannot create a V3 table. + + Recorded as a failure rather than a skip: the engine was asked and refused, + which is evidence, so the matrix cell can be contradicted by it. + """ + r = TestResult(feature_id, feature_name, version) + can_create, evidence = _v3_creation_refused() + if can_create: + r.result = "skip" + r.details = ( + "A V3 table was created unexpectedly; this feature needs a real test " + "rather than the shared V3 rejection path" + ) + return r + r.result = "fail" + r.details = ( + "Redshift is an Iceberg v2 engine: creating a format-version 3 table is " + f"rejected, so this V3 feature cannot exist here ({evidence})" + ) + return r + + +_V3_CREATION = None + + +def _v3_creation_refused() -> tuple: + """Whether this Redshift can create a format-version 3 table, measured once. + + Around twenty V3 cells share this answer, and it is a property of the engine + rather than of any one feature, so asking once keeps them consistent and keeps + the run short: the refusal is normally instant but was occasionally observed + taking the full statement timeout, which turned a cheap check into minutes of + dead waiting repeated per feature. + + Returns (created_successfully, evidence). + """ + global _V3_CREATION + if _V3_CREATION is None: + tbl = _unique("v3probe") + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (id BIGINT) USING ICEBERG {_loc(tbl)} + TABLE PROPERTIES ('format-version'='3')""" + ], timeout=FIXTURE_WRITE_TIMEOUT) + if ok: + _run_sql([f"DROP TABLE {_table(tbl)}"]) + _V3_CREATION = (True, "a format-version 3 table was created") + else: + _V3_CREATION = (False, _error_reason(out, 150)) + return _V3_CREATION + + +def _needs_external_engine(feature_id: str, feature_name: str, version: str, + what: str) -> TestResult: + """Honest skip for something no Redshift SQL surface can express.""" + r = TestResult(feature_id, feature_name, version) + r.result = "skip" + r.details = f"Not exercised: {what}" + return r + + +# --------------------------------------------------------------------------- +# Spark-created fixtures: telling "cannot create" from "cannot read" +# --------------------------------------------------------------------------- + +def _fixture(name: str) -> str: + return f"{FIXTURE_SCHEMA}.{name}" + + +def _fixture_available() -> bool: + return bool(FIXTURE_DB) + + +def _read_fixture(feature_id: str, feature_name: str, version: str, + table: str, what: str, select: str = None, + expect: str = None, forbid: str = None, + write_probe: str = None) -> TestResult: + """Read a table Spark made that Redshift itself cannot create. + + The point is to split one question into two. Redshift refusing to *create* a + v3 table says nothing about whether it can *read* one, and those are different + matrix answers: a feature it reads but cannot produce is partial support, not + absent support. + + Outcomes: + * fixtures not configured -> skip; nothing was measured + * the fixture is missing from the manifest -> skip; Spark could not build it + either, so there is nothing to read and no claim to make + * read fails -> fail; the feature is genuinely absent, not merely unwritable + * read succeeds and a write probe fails -> partial, with both halves quoted + * read succeeds and no write probe was asked for -> pass + """ + r = TestResult(feature_id, feature_name, version) + if not _fixture_available(): + r.result = "skip" + r.details = ( + f"Not measured: {what} needs a table Redshift cannot create, and no " + "Spark fixture database was configured for this run " + "(set REDSHIFT_FIXTURE_DB)" + ) + return r + + sql = select or f"SELECT COUNT(*) FROM {_fixture(table)}" + ok, out = _run_sql([sql]) + if not ok: + reason = _error_reason(out, 170) + if "not found" in reason.lower() or "does not exist" in reason.lower(): + r.result = "skip" + r.details = ( + f"Not measured: the {table} fixture is absent, so Spark could not " + f"build it either and there is nothing to read ({reason})" + ) + return r + r.result = "fail" + r.details = ( + f"Redshift cannot read {what} even when Spark creates it, so this is " + f"absent rather than merely unwritable ({reason})" + ) + return r + + if expect is not None and expect not in out: + r.result = "fail" + r.details = ( + f"The {table} fixture was readable but returned the wrong data: " + f"expected {expect!r}, got {out[:120]!r}" + ) + return r + # A negative assertion, for the cases where the interesting evidence is a row + # that must be *gone*. A reader that silently ignored a delete file would + # otherwise pass on row count alone. + if forbid is not None and forbid in out: + r.result = "fail" + r.details = ( + f"The {table} fixture was readable but {what} was not applied: " + f"{forbid!r} should have been absent, got {out[:120]!r}" + ) + return r + + read_note = f"Redshift reads {what} from a Spark-created table" + if out: + read_note += f" ({out.splitlines()[0][:80]})" + + if not write_probe: + r.result = "pass" + r.details = read_note + return r + + # Bounded well below the statement timeout. A refusal comes back in under a + # second, so anything slower is Redshift planning work it will not finish, and + # waiting the full timeout would add minutes per fixture for no extra + # information. + wrote, werr = _run_sql([write_probe], timeout=FIXTURE_WRITE_TIMEOUT) + if wrote: + r.result = "pass" + r.details = f"{read_note}, and can write to it as well" + return r + r.result = "partial" + r.details = f"{read_note}, but cannot write it: {_error_reason(werr, 150)}" + return r + + +_REST_CLAUSE_KNOWN = None + + +def _rest_clause_recognised() -> tuple: + """Whether this Redshift version has an Iceberg REST catalog client. + + Redshift reaches Iceberg through the Glue Data Catalog and nothing else, so + every non-Glue catalog cell has the same answer. That answer is a property of + Redshift's SQL grammar rather than of any endpoint, so it is measured once + per run instead of by pointing seven probes at seven dead ports, which would + add latency and no information. + + REST is the clause worth measuring because it is the one that could plausibly + appear in a future Redshift release, and because Nessie, Polaris, Unity and + Horizon are all reached over Iceberg REST. If AWS ships a REST client, this + probe notices and those cells stop being reported as absent on their own. + + Detecting the refusal needs care. Redshift does not reject an unknown + FROM clause: it discards the clause, falls back to a Data Catalog + definition and reports + + ERROR: DATABASE is mandatory for Data Catalog external schema definition. + + even when DATABASE *was* supplied. Measured against a deliberately + meaningless clause, FROM ICEBERG REST CATALOG gives byte-identical output, + while FROM HIVE METASTORE names itself and FROM DATA CATALOG succeeds. So the + test is not a string match on the error but a comparison against a clause + known to be nonsense: if the two are indistinguishable, Redshift never + understood the REST clause at all. + + Returns (recognised, evidence). + """ + global _REST_CLAUSE_KNOWN + if _REST_CLAUSE_KNOWN is not None: + return _REST_CLAUSE_KNOWN + + def attempt(name: str, clause: str) -> tuple: + schema = f"probe{name}_{RUN_TAG}" + ok, out = _run_sql([f"CREATE EXTERNAL SCHEMA {schema} {clause}"]) + if ok: + _run_sql([f"DROP SCHEMA IF EXISTS {schema}"]) + return ok, _error_reason(out, 150) + + nonsense_ok, nonsense_err = attempt( + "ctl", + f"FROM TOTALLY MADE UP CATALOG DATABASE 'ghost_{RUN_TAG}' " + f"IAM_ROLE '{ROLE_ARN}'", + ) + rest_ok, rest_err = attempt( + "rest", + f"FROM ICEBERG REST CATALOG DATABASE 'ghost_{RUN_TAG}' " + f"URI 'http://127.0.0.1:8181' IAM_ROLE '{ROLE_ARN}'", + ) + + if nonsense_ok: + # The baseline is worthless if Redshift accepts nonsense, so claim + # nothing from the comparison. + _REST_CLAUSE_KNOWN = ( + None, + "the meaningless-clause control was accepted, so no conclusion can be " + "drawn from comparing against it", + ) + elif rest_ok: + _REST_CLAUSE_KNOWN = ( + True, + "an Iceberg REST external schema was accepted, though lazily and " + "without contacting the endpoint", + ) + elif rest_err == nonsense_err: + _REST_CLAUSE_KNOWN = ( + False, + "FROM ICEBERG REST CATALOG is indistinguishable from a deliberately " + f"meaningless catalog clause tested in this run ({rest_err})", + ) + else: + _REST_CLAUSE_KNOWN = ( + True, + f"the REST clause was understood and failed on its own terms " + f"({rest_err})", + ) + return _REST_CLAUSE_KNOWN + + +def _rest_backed_catalog(feature_id: str, feature_name: str, version: str, + what: str) -> TestResult: + """A catalog that is reached over Iceberg REST, which Redshift has no client for.""" + r = TestResult(feature_id, feature_name, version) + recognised, evidence = _rest_clause_recognised() + if recognised is False: + r.result = "fail" + r.details = ( + f"Redshift reads Iceberg only through the Glue Data Catalog, and {what} " + f"is reached over Iceberg REST, which this version has no client for: " + f"{evidence}" + ) + else: + r.result = "skip" + r.details = ( + f"{what} is reached over Iceberg REST; this run could not establish " + f"that Redshift lacks a REST client, and no live endpoint was " + f"available to read a table through, so support is unverified: " + f"{evidence}" + ) + return r + + +def _non_glue_catalog(feature_id: str, feature_name: str, version: str, + what: str, why: str) -> TestResult: + """A catalog with no Redshift external-schema form at all. + + Recorded as a failure rather than a skip because Redshift's Iceberg access + goes through the Glue Data Catalog by design, so this is a real property of + the engine and not something an absent endpoint left unmeasured. + """ + r = TestResult(feature_id, feature_name, version) + r.result = "fail" + r.details = ( + f"Redshift reads Iceberg only through the Glue Data Catalog; {what} {why}" + ) + return r + + +# --------------------------------------------------------------------------- +# Core DDL / read / write +# --------------------------------------------------------------------------- + +def test_table_creation(version: str) -> TestResult: + r = TestResult("table-creation", "Table Creation", version) + if version == "v3": + return _v3_unsupported("table-creation", "Table Creation", version) + tbl = _unique("create") + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR, amount DECIMAL(9,2)) + USING ICEBERG {_loc(tbl)}""", + f"SHOW TABLE {_table(tbl)}", + ]) + if ok and "USING ICEBERG" in out: + r.result = "pass" + fmt = re.search(r"'format-version'='(\d)'", out) + r.details = ( + "CREATE TABLE ... USING ICEBERG accepted and SHOW TABLE reports it back" + + (f" at format-version {fmt.group(1)}" if fmt else "") + ) + elif ok: + r.result = "pass" + r.details = "CREATE TABLE ... USING ICEBERG accepted" + else: + r.result = "fail" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_read_support(version: str) -> TestResult: + r = TestResult("read-support", "Read Support", version) + if version == "v3": + # The cell that matters most for v3. Redshift refuses to create a v3 + # table, but reading one is a separate capability, so a Spark-built + # fixture is read and then written to. Read-yes/write-no is partial + # support, and recording it as a flat failure would be wrong. + return _read_fixture( + "read-support", "Read Support", version, FX_V3_BASIC, + "a format-version 3 table", + select=f"SELECT id, name FROM {_fixture(FX_V3_BASIC)} ORDER BY id", + expect="alpha", + write_probe=(f"INSERT INTO {_fixture(FX_V3_BASIC)} " + f"VALUES (999,'from-redshift')"), + ) + tbl = _unique("read") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1,'a'),(2,'b'),(3,'c')", + f"SELECT COUNT(*) FROM {_table(tbl)}", + ]) + if ok and "3" in out: + r.result = "pass" + r.details = "Rows written and read back through the external schema" + else: + r.result = "fail" if ok else "error" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_write_insert(version: str) -> TestResult: + r = TestResult("write-insert", "Write (INSERT)", version) + if version == "v3": + return _v3_unsupported("write-insert", "Write (INSERT)", version) + tbl = _unique("ins") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1,'a'),(2,'b')", + f"INSERT INTO {_table(tbl)} SELECT 3, 'c'", + f"SELECT COUNT(*) FROM {_table(tbl)}", + ]) + if ok and "3" in out: + r.result = "pass" + r.details = "INSERT ... VALUES and INSERT ... SELECT both write Iceberg data" + else: + r.result = "fail" if ok else "error" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_write_merge_update_delete(version: str) -> TestResult: + r = TestResult("write-merge-update-delete", "Write (MERGE/UPDATE/DELETE)", version) + if version == "v3": + return _v3_unsupported("write-merge-update-delete", + "Write (MERGE/UPDATE/DELETE)", version) + tbl = _unique("dml") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1,'first'),(2,'second'),(3,'third')", + f"UPDATE {_table(tbl)} SET name='updated' WHERE id=1", + f"DELETE FROM {_table(tbl)} WHERE id=3", + f"""MERGE INTO {_table(tbl)} USING (SELECT 2 AS id, 'merged' AS nm) src + ON {_table(tbl)}.id = src.id + WHEN MATCHED THEN UPDATE SET name = src.nm + WHEN NOT MATCHED THEN INSERT (id, name) VALUES (src.id, src.nm)""", + f"SELECT id, name FROM {_table(tbl)} ORDER BY id", + ]) + if ok and "updated" in out and "merged" in out and "third" not in out: + r.result = "pass" + r.details = "UPDATE, DELETE and MERGE all applied; final rows verified" + elif ok: + r.result = "fail" + r.details = f"DML ran but the rows are wrong: {out[:200]}" + else: + r.result = "fail" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_catalog_integration(version: str) -> TestResult: + r = TestResult("catalog-integration", "Catalog Integration", version) + if version == "v3": + return _v3_unsupported("catalog-integration", "Catalog Integration", version) + tbl = _unique("cat") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1),(2)", + f"SELECT COUNT(*) FROM {_table(tbl)}", + f"DROP TABLE {_table(tbl)}", + ]) + if ok and "2" in out: + r.result = "pass" + r.details = ( + "Full create/write/read/drop round-trip through a Glue Data Catalog " + "external schema" + ) + else: + r.result = "fail" if ok else "error" + r.details = _error_reason(out) + return r + + +# --------------------------------------------------------------------------- +# Row-level operations +# --------------------------------------------------------------------------- + +def _delete_file_evidence(tbl: str) -> tuple: + """The newest Iceberg snapshot summary for a table, as (summary, note). + + Redshift exposes no Iceberg metadata tables, so the evidence comes from the + table's own metadata.json. Iceberg records the delete bookkeeping there + directly -- total-data-files, total-delete-files, total-position-deletes, + total-equality-deletes and the added-* counters for the latest commit -- so + the copy-on-write versus merge-on-read question is answered by Iceberg itself + rather than inferred. + + This replaced a check that counted objects whose basename contained + "delete". That convention is Spark's, not Iceberg's, and Redshift does not + follow it: it writes position-delete files into data/ under ordinary names. + The name-based count therefore saw zero deletes and reported copy-on-write + for an engine that demonstrably writes position deletes, which was enough to + contradict four matrix cells in the wrong direction. + """ + try: + s3 = _client("s3") + if STORAGE_MODE == "s3tables": + # S3 Tables keeps its storage service-managed, so the metadata is + # found through the API rather than by listing a known prefix. The + # object it points at is readable with ordinary GetObject under this + # role, so the same evidence is available in both modes. + if not TABLE_BUCKET_ARN: + return None, "AWS_TABLE_BUCKET_ARN is not set" + loc = _client("s3tables").get_table_metadata_location( + tableBucketARN=TABLE_BUCKET_ARN, + namespace=S3T_NAMESPACE, + name=tbl, + ).get("metadataLocation", "") + if not loc.startswith("s3://"): + return None, f"no metadata location returned for {tbl}" + bucket, key = loc[5:].split("/", 1) + else: + if not DATA_BUCKET: + return None, "AWS_DATA_BUCKET is not set" + prefix = f"redshift/{RUN_TAG}/{tbl}/" + keys = [] + token = None + while True: + kwargs = {"Bucket": DATA_BUCKET, "Prefix": prefix, "MaxKeys": 1000} + if token: + kwargs["ContinuationToken"] = token + page = s3.list_objects_v2(**kwargs) + keys.extend(o["Key"] for o in page.get("Contents", [])) + token = page.get("NextContinuationToken") + if not token: + break + metadata = sorted(k for k in keys if k.endswith(".metadata.json")) + if not metadata: + return None, f"no metadata.json found under {prefix}" + bucket, key = DATA_BUCKET, metadata[-1] + doc = json.loads(s3.get_object(Bucket=bucket, Key=key)["Body"].read()) + except Exception as e: # noqa: BLE001 + return None, f"could not read table metadata: {type(e).__name__}: {e}" + + snapshots = doc.get("snapshots", []) + if not snapshots: + return None, "table metadata carries no snapshots" + summary = snapshots[-1].get("summary", {}) + + def count(key: str) -> int: + try: + return int(summary.get(key, 0)) + except (TypeError, ValueError): + return 0 + + note = ( + f"operation={summary.get('operation', '?')}, " + f"data-files={count('total-data-files')}, " + f"delete-files={count('total-delete-files')}, " + f"position-deletes={count('total-position-deletes')}, " + f"equality-deletes={count('total-equality-deletes')}" + ) + return summary, note + + +def _summary_count(summary: dict, key: str) -> int: + """Iceberg writes summary counters as strings, so read them as numbers.""" + try: + return int(summary.get(key, 0)) + except (TypeError, ValueError): + return 0 + + +def test_position_deletes(version: str) -> TestResult: + r = TestResult("position-deletes", "Position Deletes", version) + if version == "v3": + return _v3_unsupported("position-deletes", "Position Deletes", version) + tbl = _unique("posdel") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1,'a'),(2,'b'),(3,'c')", + f"DELETE FROM {_table(tbl)} WHERE id=2", + f"SELECT COUNT(*) FROM {_table(tbl)}", + ]) + if not ok: + r.result = "fail" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + summary, note = _delete_file_evidence(tbl) + if summary is None: + r.result = "skip" + r.details = ( + f"DELETE succeeded and the row count dropped, but whether Redshift " + f"wrote position deletes or rewrote data files is not observable: {note}" + ) + elif _summary_count(summary, "total-position-deletes") > 0: + r.result = "pass" + r.details = ( + f"DELETE wrote position deletes and left the data files in place ({note})" + ) + else: + r.result = "fail" + r.details = ( + "DELETE recorded no position deletes in the Iceberg snapshot summary " + f"({note}), so the row was removed by rewriting data" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_equality_deletes(version: str) -> TestResult: + r = TestResult("equality-deletes", "Equality Deletes", version) + if version == "v3": + return _v3_unsupported("equality-deletes", "Equality Deletes", version) + return _needs_external_engine( + "equality-deletes", "Equality Deletes", version, + "Redshift SQL has no surface that requests equality deletes: DELETE and " + "MERGE choose their own delete strategy and no table property selects it", + ) + + +def test_merge_on_read(version: str) -> TestResult: + r = TestResult("merge-on-read", "Merge-on-Read", version) + if version == "v3": + return _v3_unsupported("merge-on-read", "Merge-on-Read", version) + tbl = _unique("mor") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1,'a'),(2,'b'),(3,'c')", + f"UPDATE {_table(tbl)} SET name='changed' WHERE id=2", + f"SELECT name FROM {_table(tbl)} WHERE id=2", + ]) + if not ok: + r.result = "fail" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + summary, note = _delete_file_evidence(tbl) + if summary is None: + r.result = "skip" + r.details = f"UPDATE applied, but the write mode is not observable: {note}" + elif _summary_count(summary, "total-delete-files") > 0: + r.result = "pass" + r.details = ( + "UPDATE committed delete files rather than rewriting the table, which " + f"is merge-on-read ({note})" + ) + else: + r.result = "fail" + r.details = ( + f"UPDATE committed no delete files ({note}); Redshift resolved it by " + "rewriting data, which is copy-on-write rather than merge-on-read" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_copy_on_write(version: str) -> TestResult: + r = TestResult("copy-on-write", "Copy-on-Write", version) + if version == "v3": + return _v3_unsupported("copy-on-write", "Copy-on-Write", version) + # Redshift's own default is merge-on-read, as the delete counters show, so + # the question this cell asks is whether copy-on-write can be selected at + # all. Iceberg's knob is write.delete.mode / write.update.mode, so the test + # asks for it and then checks whether the next UPDATE actually honoured it. + tbl = _unique("cow") + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG + {_loc(tbl)} TABLE PROPERTIES ( + 'write.delete.mode'='copy-on-write', + 'write.update.mode'='copy-on-write')""", + ]) + if not ok: + r.result = "fail" + r.details = ( + "Copy-on-write cannot be selected: Redshift rejects the Iceberg " + "write.delete.mode / write.update.mode properties, and its own " + f"default is merge-on-read ({_error_reason(out, 150)})" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + ok, out = _run_sql([ + f"INSERT INTO {_table(tbl)} VALUES (1,'a'),(2,'b')", + f"UPDATE {_table(tbl)} SET name='z' WHERE id=1", + f"SELECT COUNT(*) FROM {_table(tbl)}", + ]) + if not ok: + r.result = "fail" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + summary, note = _delete_file_evidence(tbl) + if summary is None: + r.result = "skip" + r.details = ( + "The copy-on-write properties were accepted, but whether the UPDATE " + f"honoured them is not observable: {note}" + ) + elif _summary_count(summary, "total-delete-files") == 0: + r.result = "pass" + r.details = ( + "With write.update.mode=copy-on-write the UPDATE rewrote data files " + f"and committed no delete files ({note})" + ) + else: + r.result = "fail" + r.details = ( + "Redshift accepted write.delete.mode / write.update.mode = " + "copy-on-write but ignored them: the UPDATE still committed delete " + f"files, so it stayed merge-on-read ({note})" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_deletion_vectors(version: str) -> TestResult: + if version == "v2": + r = TestResult("deletion-vectors", "Deletion Vectors", version) + r.result = "skip" + r.details = "V3-only feature; not applicable to format-version 2 tables" + return r + # The fixture is a v3 table where Spark deleted id=2 through a deletion + # vector. Asserting the surviving ids rather than just a row count is the + # point: a reader that ignored the vector would return the deleted row, and a + # count alone could coincide for other reasons. + return _read_fixture( + "deletion-vectors", "Deletion Vectors", version, FX_V3_DV, + "a v3 table carrying deletion vectors", + # Spark inserted 1..4 then deleted 2 via a deletion vector, so a correct + # reader returns 1, 3, 4 and never 2. + select=(f"SELECT LISTAGG(id, ',') WITHIN GROUP (ORDER BY id) " + f"FROM {_fixture(FX_V3_DV)}"), + expect="1,3,4", + forbid="2,", + # INSERT rather than DELETE on purpose. Either would be refused, but + # Redshift answers an INSERT immediately with the version error, whereas a + # DELETE against a deletion-vector table was observed to hang until the + # statement timeout, which is slower and much weaker evidence. + write_probe=f"INSERT INTO {_fixture(FX_V3_DV)} VALUES (99,'from-redshift')", + ) + + +# --------------------------------------------------------------------------- +# Schema and table management +# --------------------------------------------------------------------------- + +def test_schema_evolution(version: str) -> TestResult: + r = TestResult("schema-evolution", "Schema Evolution", version) + if version == "v3": + return _v3_unsupported("schema-evolution", "Schema Evolution", version) + tbl = _unique("evo") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1,'a')", + f"ALTER TABLE {_table(tbl)} ADD COLUMN added VARCHAR", + f"ALTER TABLE {_table(tbl)} RENAME COLUMN added TO renamed", + f"ALTER TABLE {_table(tbl)} DROP COLUMN renamed", + f"SELECT COUNT(*) FROM {_table(tbl)}", + ]) + if ok: + r.result = "pass" + r.details = ( + "ADD, RENAME and DROP COLUMN all accepted as metadata-only changes, " + "with existing rows still readable" + ) + else: + r.result = "fail" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_type_promotion(version: str) -> TestResult: + r = TestResult("type-promotion", "Type Promotion / Widening", version) + if version == "v3": + return _v3_unsupported("type-promotion", "Type Promotion / Widening", version) + tbl = _unique("prom") + # Start from INT and FLOAT4 so there is somewhere to widen to. Iceberg allows + # int -> bigint and float -> double; narrowing is rejected by design. + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (small_id INT, ratio FLOAT4, name VARCHAR) + USING ICEBERG {_loc(tbl)}""", + f"INSERT INTO {_table(tbl)} VALUES (1, 1.5, 'a')", + f"ALTER TABLE {_table(tbl)} ALTER COLUMN small_id TYPE BIGINT", + f"ALTER TABLE {_table(tbl)} ALTER COLUMN ratio TYPE FLOAT8", + f"SELECT small_id, name FROM {_table(tbl)}", + ]) + if not ok: + r.result = "fail" + r.details = f"Widening rejected: {_error_reason(out)}" + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + # Narrowing must be refused; if it were allowed that is a correctness problem + # worth recording rather than a pass. + narrow_ok, narrow_out = _run_sql([ + f"ALTER TABLE {_table(tbl)} ALTER COLUMN small_id TYPE INT" + ]) + if narrow_ok: + r.result = "pass" + r.details = ( + "int->bigint and float->double both accepted; narrowing back was also " + "accepted, which the Iceberg spec does not allow" + ) + else: + r.result = "pass" + r.details = ( + "int->bigint and float->double accepted as metadata-only widenings, " + f"and narrowing is refused ({_error_reason(narrow_out, 90)})" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_column_default_values(version: str) -> TestResult: + if version == "v2": + r = TestResult("column-default-values", "Column Default Values", version) + r.result = "skip" + r.details = "V3-only feature; not applicable to format-version 2 tables" + return r + return _v3_unsupported("column-default-values", "Column Default Values", version) + + +def test_time_travel(version: str) -> TestResult: + r = TestResult("time-travel", "Time Travel / Snapshots", version) + if version == "v3": + return _v3_unsupported("time-travel", "Time Travel / Snapshots", version) + tbl = _unique("tt") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1)", + f"INSERT INTO {_table(tbl)} VALUES (2)", + ]) + if not ok: + r.result = "error" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + # Try every syntax an engine might accept, so "not supported" is a measured + # conclusion rather than one failed guess. + attempts = { + "FOR SYSTEM_TIME AS OF": f"SELECT COUNT(*) FROM {_table(tbl)} FOR SYSTEM_TIME AS OF '2026-01-01'", + "FOR TIMESTAMP AS OF": f"SELECT COUNT(*) FROM {_table(tbl)} FOR TIMESTAMP AS OF '2026-01-01'", + "FOR VERSION AS OF": f"SELECT COUNT(*) FROM {_table(tbl)} FOR VERSION AS OF 1", + "FOR SYSTEM_VERSION AS OF": f"SELECT COUNT(*) FROM {_table(tbl)} FOR SYSTEM_VERSION AS OF 1", + "$snapshots metadata table": f"SELECT COUNT(*) FROM {_table(tbl)}$snapshots", + } + accepted = [] + reasons = [] + for label, sql in attempts.items(): + a_ok, a_out = _run_sql([sql]) + if a_ok: + accepted.append(label) + else: + reasons.append(f"{label}: {_error_reason(a_out, 70)}") + + if accepted: + r.result = "pass" + r.details = f"Time travel available via {', '.join(accepted)}" + else: + r.result = "fail" + r.details = ( + "No time-travel syntax is accepted and no snapshot metadata table is " + f"exposed. Tried {len(attempts)} forms; first: {reasons[0] if reasons else 'n/a'}" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_table_maintenance(version: str) -> TestResult: + r = TestResult("table-maintenance", "Table Maintenance", version) + if version == "v3": + return _v3_unsupported("table-maintenance", "Table Maintenance", version) + tbl = _unique("maint") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1)", + ]) + if not ok: + r.result = "error" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + attempts = { + "VACUUM": f"VACUUM {_table(tbl)}", + "OPTIMIZE": f"OPTIMIZE TABLE {_table(tbl)}", + "ANALYZE": f"ANALYZE {_table(tbl)}", + "CALL rewrite_data_files": f"CALL system.rewrite_data_files('{_table(tbl)}')", + } + accepted = [label for label, sql in attempts.items() if _run_sql([sql])[0]] + if accepted: + r.result = "pass" + r.details = f"Maintenance available via {', '.join(accepted)}" + else: + r.result = "fail" + r.details = ( + "No maintenance statement is accepted for Iceberg tables: compaction " + "and snapshot expiry are left to Glue or S3 Tables rather than Redshift" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_branching_tagging(version: str) -> TestResult: + r = TestResult("branching-tagging", "Branching & Tagging", version) + if version == "v3": + return _v3_unsupported("branching-tagging", "Branching & Tagging", version) + tbl = _unique("branch") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1)", + ]) + if not ok: + r.result = "error" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + attempts = [ + f"ALTER TABLE {_table(tbl)} CREATE BRANCH test_branch", + f"ALTER TABLE {_table(tbl)} CREATE TAG test_tag", + f"SELECT COUNT(*) FROM {_table(tbl)} VERSION AS OF 'test_branch'", + ] + if any(_run_sql([sql])[0] for sql in attempts): + r.result = "pass" + r.details = "A branch or tag statement was accepted" + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + _run_sql([f"DROP TABLE {_table(tbl)}"]) + + # Redshift has no branch or tag DDL, which only settles the write half. The + # fixture is a Spark table with a branch, a tag, and a row that exists only on + # the branch, so it can also answer whether an existing ref is addressable. + if not _fixture_available(): + r.result = "fail" + r.details = ( + "No branch or tag DDL is accepted: Redshift exposes only the table's " + "current state. Whether it can read a ref another engine created was " + "not measured (no Spark fixture configured)" + ) + return r + + ok, out = _run_sql([f"SELECT COUNT(*) FROM {_fixture(FX_V2_BRANCH)}"]) + if not ok: + r.result = "fail" + r.details = ( + "No branch or tag DDL is accepted, and the branched fixture table is " + f"not readable either ({_error_reason(out, 130)})" + ) + return r + + # Every spelling of a ref that Redshift might plausibly accept. Names resolve + # through Glue, which has no concept of a ref, so all are expected to miss. + ref_forms = { + "VERSION AS OF": (f"SELECT COUNT(*) FROM {_fixture(FX_V2_BRANCH)} " + f"VERSION AS OF 'audit_branch'"), + "table@branch": f'SELECT COUNT(*) FROM {FIXTURE_SCHEMA}."{FX_V2_BRANCH}@audit_branch"', + "table$branch": f'SELECT COUNT(*) FROM {FIXTURE_SCHEMA}."{FX_V2_BRANCH}$audit_branch"', + "branch_ suffix": (f'SELECT COUNT(*) FROM {FIXTURE_SCHEMA}.' + f'"{FX_V2_BRANCH}.branch_audit_branch"'), + } + reached = [name for name, sql in ref_forms.items() if _run_sql([sql])[0]] + if reached: + r.result = "partial" + r.details = ( + "Redshift cannot create branches or tags, but it can read a ref that " + f"another engine created, via {', '.join(reached)}" + ) + else: + r.result = "fail" + r.details = ( + "Redshift has no branch or tag support in either direction: no DDL is " + "accepted, and although the branched fixture table reads fine on main, " + f"none of {len(ref_forms)} ref spellings resolves, because table names " + "go through Glue and Glue has no notion of a ref" + ) + return r + + +# --------------------------------------------------------------------------- +# Partitioning +# --------------------------------------------------------------------------- + +def test_hidden_partitioning(version: str) -> TestResult: + r = TestResult("hidden-partitioning", "Hidden Partitioning", version) + if version == "v3": + return _v3_unsupported("hidden-partitioning", "Hidden Partitioning", version) + tbl = _unique("hidden") + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (id BIGINT, ts TIMESTAMP, name VARCHAR) + USING ICEBERG {_loc(tbl)} + PARTITIONED BY (day(ts), bucket(8, id))""", + f"INSERT INTO {_table(tbl)} VALUES (1, '2026-01-01 10:00:00', 'a'), (2, '2026-02-01 11:00:00', 'b')", + # The point of hidden partitioning: the query filters on the raw column, + # not on a derived partition column. + f"SELECT COUNT(*) FROM {_table(tbl)} WHERE ts >= '2026-02-01'", + f"SHOW TABLE {_table(tbl)}", + ]) + if not ok: + r.result = "fail" + r.details = _error_reason(out) + elif "PARTITIONED BY" in out.upper(): + r.result = "pass" + r.details = ( + "Transform partitioning declared with day() and bucket() and stored on " + "the table; queries filter on the source column with no derived column " + "in the schema" + ) + else: + # Accepting the clause is not the same as honouring it. In the s3tables + # mode Redshift takes PARTITIONED BY at CREATE without complaint and then + # stores an unpartitioned table, so trusting the absence of an error here + # would report partitioning that does not exist. + # + # That is still only half the question. ALTER TABLE ... ADD PARTITION + # FIELD does work there, so a transform-partitioned table is reachable, + # just not in one statement. Reporting a flat failure would overstate the + # gap, so the second route is tried before concluding. + alter_ok, alter_out = _run_sql([ + f"ALTER TABLE {_table(tbl)} ADD PARTITION FIELD day(ts)", + f"ALTER TABLE {_table(tbl)} ADD PARTITION FIELD bucket(8, id)", + f"SHOW TABLE {_table(tbl)}", + ]) + if alter_ok and "PARTITIONED BY" in alter_out.upper(): + r.result = "partial" + r.details = ( + "PARTITIONED BY at CREATE is accepted and then silently discarded " + "(SHOW TABLE reports no spec), but ALTER TABLE ADD PARTITION FIELD " + "does apply day() and bucket(), so transform partitioning is " + "reachable in two statements rather than one" + ) + else: + r.result = "fail" + r.details = ( + "PARTITIONED BY was accepted at CREATE without error but silently " + "discarded: SHOW TABLE reports no partition spec, and ADD PARTITION " + f"FIELD does not establish one either ({_error_reason(alter_out, 110)})" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_partition_evolution(version: str) -> TestResult: + r = TestResult("partition-evolution", "Partition Evolution", version) + if version == "v3": + return _v3_unsupported("partition-evolution", "Partition Evolution", version) + # The first partition field is established with ALTER rather than at CREATE + # on purpose. A CREATE-time PARTITIONED BY is silently discarded in the + # s3tables mode, and depending on it here would report partition *evolution* + # as broken when what actually failed was the initial declaration. That + # declaration is hidden-partitioning's cell to report; this one is about + # changing the spec afterwards. + tbl = _unique("pevo") + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (id BIGINT, ts TIMESTAMP, name VARCHAR) + USING ICEBERG {_loc(tbl)}""", + f"INSERT INTO {_table(tbl)} VALUES (1, '2026-01-01 00:00:00', 'a')", + f"ALTER TABLE {_table(tbl)} ADD PARTITION FIELD year(ts)", + f"INSERT INTO {_table(tbl)} VALUES (2, '2026-02-01 00:00:00', 'b')", + f"SHOW TABLE {_table(tbl)}", + ]) + if not ok: + r.result = "fail" + r.details = f"ADD PARTITION FIELD failed: {_error_reason(out)}" + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + if "PARTITIONED BY" not in out.upper(): + r.result = "fail" + r.details = ( + "ADD PARTITION FIELD was accepted but no partition spec is stored, so " + "the spec cannot be evolved" + ) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + ok, out = _run_sql([ + f"ALTER TABLE {_table(tbl)} REPLACE PARTITION FIELD year(ts) WITH month(ts)", + f"ALTER TABLE {_table(tbl)} DROP PARTITION FIELD month(ts)", + # Rows written under both specs must still read back together. + f"SELECT COUNT(*) FROM {_table(tbl)}", + ]) + if ok and "2" in out: + r.result = "pass" + r.details = ( + "ADD, REPLACE and DROP PARTITION FIELD all accepted, and rows written " + "under the old and new specs read back together" + ) + elif ok: + r.result = "fail" + r.details = f"Partition spec changed but the row count is wrong: {out[:150]}" + else: + r.result = "fail" + r.details = f"Spec could not be evolved: {_error_reason(out)}" + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_multi_arg_transforms(version: str) -> TestResult: + if version == "v2": + r = TestResult("multi-arg-transforms", "Multi-Argument Transforms", version) + r.result = "skip" + r.details = "V3-only feature; not applicable to format-version 2 tables" + return r + r = TestResult("multi-arg-transforms", "Multi-Argument Transforms", version) + tbl = _unique("marg") + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (a BIGINT, b BIGINT) USING ICEBERG {_loc(tbl)} + PARTITIONED BY (bucket(4, a, b))""" + ]) + if ok: + r.result = "pass" + r.details = "A transform over more than one source column was accepted" + _run_sql([f"DROP TABLE {_table(tbl)}"]) + else: + r.result = "fail" + r.details = ( + "A transform cannot take more than one source column: " + f"{_error_reason(out, 150)}" + ) + return r + + +# --------------------------------------------------------------------------- +# Read/write extras +# --------------------------------------------------------------------------- + +def test_statistics(version: str) -> TestResult: + r = TestResult("statistics", "Statistics (Column Metrics)", version) + if version == "v3": + return _v3_unsupported("statistics", "Statistics (Column Metrics)", version) + tbl = _unique("stats") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT, name VARCHAR) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1,'a'),(2,'b'),(3,'c')", + # A predicate that should be answerable from column bounds. + f"EXPLAIN SELECT COUNT(*) FROM {_table(tbl)} WHERE id > 2", + ]) + if ok: + r.result = "pass" + r.details = ( + "Iceberg column metrics are written on insert and the planner reads " + "the table's statistics; Redshift documents using them to prune scans" + ) + else: + r.result = "fail" + r.details = _error_reason(out) + _run_sql([f"DROP TABLE {_table(tbl)}"]) + return r + + +def test_bloom_filters(version: str) -> TestResult: + r = TestResult("bloom-filters", "Bloom Filters & Puffin", version) + if version == "v3": + return _v3_unsupported("bloom-filters", "Bloom Filters & Puffin", version) + tbl = _unique("bloom") + ok, out = _run_sql([ + f"""CREATE TABLE {_table(tbl)} (id BIGINT) USING ICEBERG {_loc(tbl)} + TABLE PROPERTIES ('write.parquet.bloom-filter-enabled.column.id'='true')""" + ]) + if ok: + _run_sql([f"DROP TABLE {_table(tbl)}"]) + r.result = "skip" + r.details = ( + "The bloom-filter table property was accepted, but Redshift exposes no " + "way to confirm a filter was written, so this is not verified" + ) + else: + r.result = "fail" + r.details = ( + "Redshift accepts only 'compression_type' as an Iceberg table property, " + f"so bloom filters cannot be requested: {_error_reason(out, 140)}" + ) + return r + + +# --------------------------------------------------------------------------- +# Catalogs +# --------------------------------------------------------------------------- + +def test_aws_glue_catalog(version: str) -> TestResult: + r = TestResult("aws-glue-catalog", "AWS Glue Catalog", version) + if version == "v3": + return _v3_unsupported("aws-glue-catalog", "AWS Glue Catalog", version) + tbl = _unique("glue") + ok, out = _run_sql([ + f"CREATE TABLE {_table(tbl)} (id BIGINT) USING ICEBERG {_loc(tbl)}", + f"INSERT INTO {_table(tbl)} VALUES (1)", + f"SELECT COUNT(*) FROM {_table(tbl)}", + f"DROP TABLE {_table(tbl)}", + ]) + if ok and "1" in out: + r.result = "pass" + r.details = ( + "The Glue Data Catalog is the catalog Redshift uses for Iceberg: table " + "created, written, read and dropped through it" + ) + else: + r.result = "fail" if ok else "error" + r.details = _error_reason(out) + return r + + +def test_rest_catalog(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("rest-catalog", "REST Catalog", version) + # Redshift's own tables can be published *into* Glue's Iceberg REST + # endpoint, but that is Redshift as a producer. This cell asks whether + # Redshift can consume a REST catalog as a client. + return _rest_backed_catalog("rest-catalog", "REST Catalog", version, + "a generic Iceberg REST catalog") + + +def test_hive_metastore(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("hive-metastore", "Hive Metastore", version) + # Redshift does have a FROM HIVE METASTORE clause, and it is the one non-Glue + # source it recognises: measured in this account it creates the schema even + # against a dead endpoint, whereas the REST clause is indistinguishable from + # nonsense. That clause is for Hive external tables though, and Redshift's + # Iceberg reader is bound to the Data Catalog, so it does not make this a + # route to Iceberg. Proving that either way needs a live metastore, so the + # detail says exactly how far the evidence goes. + return _non_glue_catalog( + "hive-metastore", "Hive Metastore", version, + "a Hive Metastore holding Iceberg tables", + "is not such a route: FROM HIVE METASTORE is recognised, but it registers " + "Hive external tables rather than Iceberg ones, and this was not verified " + "against a live metastore", + ) + + +def test_hadoop_catalog(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("hadoop-catalog", "Hadoop Catalog", version) + return _non_glue_catalog( + "hadoop-catalog", "Hadoop Catalog", version, + "a filesystem (Hadoop) Iceberg catalog", + "has no external-schema form, since a warehouse path on S3 cannot be " + "registered as a catalog without Glue metadata behind it", + ) + + +def test_jdbc_catalog(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("jdbc-catalog", "JDBC Catalog", version) + return _non_glue_catalog( + "jdbc-catalog", "JDBC Catalog", version, + "a JDBC-backed Iceberg catalog", + "has no external-schema form; FROM POSTGRES and FROM MYSQL exist for " + "federated query against those databases, not for reading an Iceberg " + "catalog stored in them", + ) + + +def test_nessie(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("nessie", "Nessie", version) + return _rest_backed_catalog("nessie", "Nessie", version, "Nessie") + + +def test_polaris(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("polaris", "Polaris", version) + return _rest_backed_catalog("polaris", "Polaris", version, + "Apache Polaris") + + +def test_unity_catalog(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("unity-catalog", "Unity Catalog", version) + return _rest_backed_catalog("unity-catalog", "Unity Catalog", version, + "Databricks Unity Catalog") + + +def test_snowflake_horizon_catalog(version: str) -> TestResult: + if version == "v3": + return _v3_unsupported("snowflake-horizon-catalog", + "Snowflake Horizon Catalog", version) + return _rest_backed_catalog("snowflake-horizon-catalog", + "Snowflake Horizon Catalog", version, + "Snowflake Horizon") + + +# --------------------------------------------------------------------------- +# V3 data types and advanced features +# --------------------------------------------------------------------------- + +def _v3_type(feature_id: str, feature_name: str, version: str, + column_sql: str, fixture: str = "") -> TestResult: + """A V3 type: try it on a V2 table too, so the reason is precise. + + A type can fail either because the engine has no such type at all, or only + because it needs a V3 table the engine cannot create. Distinguishing them + keeps the recorded reason honest. + """ + if version == "v2": + r = TestResult(feature_id, feature_name, version) + r.result = "skip" + r.details = "V3-only feature; not applicable to format-version 2 tables" + return r + + r = TestResult(feature_id, feature_name, version) + tbl = _unique("t") + v2_ok, v2_out = _run_sql([ + f"CREATE TABLE {_table(tbl)} ({column_sql}) USING ICEBERG {_loc(tbl)}" + ]) + if v2_ok: + _run_sql([f"DROP TABLE {_table(tbl)}"]) + r.result = "fail" + r.details = ( + "The column type is accepted, but only on a format-version 2 table; " + "Redshift cannot create the V3 table this feature requires" + ) + return r + + # Redshift cannot declare the type. That still leaves open whether it can + # read one, which is a different cell, so ask a Spark-built fixture. Reading + # is the only thing on offer here: there is no write probe, because Redshift + # refuses to write any v3 table regardless of the column types in it. + if fixture: + probe = _read_fixture(feature_id, feature_name, version, fixture, + f"the {feature_name} type") + if probe.result != "skip": + probe.details += ( + f"; Redshift cannot declare the type itself " + f"({_error_reason(v2_out, 90)})" + ) + return probe + # Fall through to the DDL-only verdict, but keep the fixture's reason so + # the report says which half went unmeasured. + r.result = "fail" + r.details = ( + f"Type not available in Redshift Iceberg DDL: " + f"{_error_reason(v2_out, 110)}. Read support unmeasured: " + f"{probe.details[:120]}" + ) + return r + + r.result = "fail" + r.details = ( + f"Type not available in Redshift Iceberg DDL: {_error_reason(v2_out, 150)}" + ) + return r + + +def test_variant_type(version: str) -> TestResult: + return _v3_type("variant-type", "Variant Type", version, + "id BIGINT, v VARIANT", FX_V3_VARIANT) + + +def test_shredded_variant(version: str) -> TestResult: + if version == "v2": + r = TestResult("shredded-variant", "Shredded Variant", version) + r.result = "skip" + r.details = "V3-only feature; not applicable to format-version 2 tables" + return r + # Shredding is a physical layout choice inside a variant column, so it can + # only matter to an engine that can read a variant at all. The variant + # fixture answers that, and it is the strongest statement available here. + probe = _read_fixture("shredded-variant", "Shredded Variant", version, + FX_V3_VARIANT, "a variant column") + if probe.result == "fail": + probe.details = ( + "Redshift cannot read a variant column at all, so shredded variants " + f"cannot be read either: {probe.details}" + ) + elif probe.result in ("pass", "partial"): + # Reading a variant does not prove the shredded encoding was exercised. + probe.result = "skip" + probe.details = ( + "A variant column is readable, but whether it was shredded is not " + "observable from Redshift, so shredding itself is unverified" + ) + return probe + + +def test_geometry_type(version: str) -> TestResult: + return _v3_type("geometry-type", "Geometry / Geo Types", version, + "id BIGINT, g GEOMETRY", FX_V3_GEOMETRY) + + +def test_nanosecond_timestamps(version: str) -> TestResult: + return _v3_type("nanosecond-timestamps", "Nanosecond Timestamps", version, + "id BIGINT, ts TIMESTAMP_NS", FX_V3_TS_NS) + + +def test_lineage(version: str) -> TestResult: + if version == "v2": + r = TestResult("lineage", "Lineage Tracking", version) + r.result = "skip" + r.details = "V3-only feature; not applicable to format-version 2 tables" + return r + return _v3_unsupported("lineage", "Lineage Tracking", version) + + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- + +ALL_TESTS = [ + test_table_creation, + test_read_support, + test_write_insert, + test_write_merge_update_delete, + test_catalog_integration, + test_position_deletes, + test_equality_deletes, + test_merge_on_read, + test_copy_on_write, + test_deletion_vectors, + test_schema_evolution, + test_type_promotion, + test_column_default_values, + test_time_travel, + test_table_maintenance, + test_branching_tagging, + test_hidden_partitioning, + test_partition_evolution, + test_multi_arg_transforms, + test_statistics, + test_bloom_filters, + test_aws_glue_catalog, + test_rest_catalog, + test_hive_metastore, + test_hadoop_catalog, + test_jdbc_catalog, + test_nessie, + test_polaris, + test_unity_catalog, + test_snowflake_horizon_catalog, + test_variant_type, + test_shredded_variant, + test_geometry_type, + test_nanosecond_timestamps, + test_lineage, +] + + +# --------------------------------------------------------------------------- +# Report generation +# --------------------------------------------------------------------------- + +def load_json_support() -> dict: + """Load the recorded support levels for the platform under test.""" + with open(os.path.join(REPO_ROOT, MATRIX_DATA_PATH)) as f: + data = json.load(f) + result = {} + for key, val in data.get("support", {}).items(): + parts = key.split(":") + if len(parts) == 3 and parts[0] == MATRIX_PLATFORM_ID: + result[(parts[1], parts[2])] = val.get("level", "unknown") + return result + + +def load_matrix_features() -> dict: + with open(os.path.join(REPO_ROOT, "src", "data", "features.json")) as f: + data = json.load(f) + return { + feat["id"]: { + "name": feat.get("name", feat["id"]), + "introducedIn": feat.get("introducedIn", "v2"), + } + for feat in data.get("features", []) + } + + +def compute_coverage(results: list) -> dict: + matrix = load_matrix_features() + tested = {r.feature_id for r in results} + uncovered = sorted(set(matrix) - tested) + return { + "matrix_feature_count": len(matrix), + "tested_feature_count": len(tested), + "uncovered": [{"id": fid, "name": matrix[fid]["name"]} for fid in uncovered], + "extra": sorted(tested - set(matrix)), + } + + +def compute_match(test_result: str, json_level: str) -> bool: + """Whether an executed test agrees with the recorded support level. + + A skip or error is not evidence either way, so it cannot disagree; those are + counted separately as "unverified". + """ + if test_result in ("skip", "error"): + return True + if test_result == "pass": + return json_level in ("full", "partial") + # An explicit partial has to meet an explicit partial. Accepting "full" or + # "none" here would make the level unfalsifiable in the one case where the + # test actually measured both halves of it. + if test_result == "partial": + return json_level == "partial" + if test_result == "fail": + return json_level in ("none", "partial") + return True + + +def generate_report(results: list) -> dict: + json_support = load_json_support() + tests_output = [] + discrepancies = 0 + unverified = 0 + for r in results: + level = json_support.get((r.feature_id, r.version_tested), "unknown") + match = compute_match(r.result, level) + if not match: + discrepancies += 1 + is_unverified = r.result in ("skip", "error") + if is_unverified: + unverified += 1 + tests_output.append({ + **r.to_dict(), + "json_level": level, + "match": match, + "verified": not is_unverified, + }) + + coverage = compute_coverage(results) + return { + "timestamp": datetime.now(tz=timezone.utc).isoformat(), + "engine": "Redshift", + "mode": STORAGE_MODE, + "redshift_version": REDSHIFT_VERSION, + "platform": MATRIX_PLATFORM_ID, + "platform_label": PLATFORM_LABEL, + "catalog_mode": CATALOG_MODE, + "versions_tested": VERSIONS, + "coverage": coverage, + "tests": tests_output, + "summary": { + "total": len(results), + "passed": sum(1 for r in results if r.result == "pass"), + "partial": sum(1 for r in results if r.result == "partial"), + "failed": sum(1 for r in results if r.result == "fail"), + "skipped": sum(1 for r in results if r.result == "skip"), + "errors": sum(1 for r in results if r.result == "error"), + "discrepancies": discrepancies, + "unverified": unverified, + "uncovered_features": len(coverage["uncovered"]), + }, + } + + +def generate_markdown(report: dict) -> str: + s = report["summary"] + lines = [ + "# Redshift Iceberg Feature Test Report", + "", + f"- **Timestamp:** {report['timestamp']}", + f"- **Redshift Version:** {report['redshift_version']}", + f"- **Storage mode:** {report['mode']}", + f"- **Catalog:** {report.get('catalog_mode', 'unknown')}", + ] + if report.get("platform_label"): + lines.append(f"- **Platform:** {report['platform_label']}") + lines += [ + f"- **Format Versions Tested:** {', '.join(report.get('versions_tested', []))}", + "", + "## Summary", + "", + "| Metric | Count |", + "|--------|-------|", + f"| Total | {s['total']} |", + f"| Passed | {s['passed']} |", + f"| Partial | {s.get('partial', 0)} |", + f"| Failed | {s['failed']} |", + f"| Skipped | {s['skipped']} |", + f"| Errors | {s['errors']} |", + f"| Discrepancies vs matrix | {s['discrepancies']} |", + f"| Unverified (skip/error) | {s['unverified']} |", + f"| Uncovered matrix features | {s.get('uncovered_features', 0)} |", + "", + "`Failed` is a result, not a defect: it records that Redshift does not " + "support the feature. `Partial` means the feature was measured as " + "half-supported, typically readable but not writable, which is a finding " + "rather than an uncertainty. A discrepancy means the observed behaviour " + "disagrees with the recorded matrix cell.", + "", + ] + + cov = report.get("coverage") + if cov: + lines.append( + f"**Matrix coverage:** {cov['tested_feature_count']}/" + f"{cov['matrix_feature_count']} features in `features.json` have a test." + ) + if cov["uncovered"]: + lines += ["", "### Uncovered matrix features (no test!)", ""] + for f in cov["uncovered"]: + lines.append(f"- **{f['name']}** (`{f['id']}`) - add a `test_*` " + "function and register it in `ALL_TESTS`") + if cov.get("extra"): + lines += ["", "> Note: tests exist for ids not in the matrix: " + f"{', '.join(cov['extra'])}"] + lines.append("") + + lines += [ + "## Test Results", + "", + "| Feature | Version | Result | Matrix | Match | Details |", + "|---------|---------|--------|--------|-------|---------|", + ] + label = {"pass": "PASS", "partial": "PARTIAL", "fail": "FAIL", + "skip": "SKIP", "error": "ERR"} + for t in report["tests"]: + details = (t["details"] or "")[:150].replace("\n", " ").replace("|", "\\|") + lines.append( + f"| {t['feature_name'].replace('|', '')} | {t['version']} " + f"| {label.get(t['result'], '?')} | {t['json_level']} " + f"| {'ok' if t['match'] else 'DISCREPANCY'} | {details} |" + ) + + discs = [t for t in report["tests"] if not t["match"]] + if discs: + lines += ["", "## Discrepancies", ""] + for t in discs: + lines.append( + f"- **{t['feature_name']}** ({t['version']}): observed " + f"`{t['result']}`, matrix says `{t['json_level']}` — " + f"{(t['details'] or '')[:300]}" + ) + + unver = [t for t in report["tests"] if not t["verified"]] + if unver: + lines += ["", "## Unverified", "", + "These could not be exercised, so they neither confirm nor " + "contradict the matrix:", ""] + for t in unver: + lines.append( + f"- **{t['feature_name']}** ({t['version']}): matrix " + f"`{t['json_level']}` — {(t['details'] or '')[:200]}" + ) + + return "\n".join(lines) + "\n" + + +# --------------------------------------------------------------------------- +# Setup and teardown +# --------------------------------------------------------------------------- + +def setup_catalog() -> tuple: + """Create the Glue database and the external schema this run writes through. + + Returns (ok, message). The external schema is what makes writes possible at + all: the auto-mounted catalog cannot write, because it authorises data access + with the caller's IAM session and a Data API connection has none. + """ + if STORAGE_MODE == "s3buckets": + try: + glue = _client("glue") + glue.create_database(DatabaseInput={"Name": GLUE_DB}) + print(f"[setup] created Glue database {GLUE_DB}") + except Exception as e: # noqa: BLE001 + if "AlreadyExists" not in type(e).__name__: + return False, f"could not create Glue database: {e}" + ok, out = _run_sql([ + f"""CREATE EXTERNAL SCHEMA {SCHEMA} + FROM DATA CATALOG DATABASE '{GLUE_DB}' + IAM_ROLE '{ROLE_ARN}'""" + ]) + if not ok: + return False, f"could not create external schema: {_error_reason(out)}" + print(f"[setup] created external schema {SCHEMA} over {GLUE_DB}") + _setup_fixture_schema() + return True, "ready" + + # s3tables. A table bucket is a *federated* Glue catalog, and Redshift cannot + # name one directly: neither "@s3tablescatalog".ns.t nor a + # CATALOG_ID holding the federated path resolves. The documented route is a + # resource link -- an ordinary database in the DEFAULT catalog whose + # TargetDatabase points at the federated namespace -- which the external + # schema then names, with CATALOG_ID set to the plain account id. + if not TABLE_BUCKET_ARN: + return False, "AWS_TABLE_BUCKET_ARN is required for the s3tables mode" + account = TABLE_BUCKET_ARN.split(":")[4] + fed_catalog_id = f"{account}:s3tablescatalog/{TABLE_BUCKET_ARN.split('/')[-1]}" + + try: + _client("s3tables").create_namespace( + tableBucketARN=TABLE_BUCKET_ARN, namespace=[S3T_NAMESPACE] + ) + print(f"[setup] created S3 Tables namespace {S3T_NAMESPACE}") + except Exception as e: # noqa: BLE001 + if "Conflict" not in type(e).__name__ and "Already" not in type(e).__name__: + return False, f"could not create S3 Tables namespace: {e}" + + try: + _client("glue").create_database( + CatalogId=account, + DatabaseInput={ + "Name": RESOURCE_LINK, + "TargetDatabase": { + "CatalogId": fed_catalog_id, + "DatabaseName": S3T_NAMESPACE, + }, + }, + ) + print(f"[setup] created Glue resource link {RESOURCE_LINK} " + f"-> {fed_catalog_id}/{S3T_NAMESPACE}") + except Exception as e: # noqa: BLE001 + if "AlreadyExists" not in type(e).__name__: + return False, f"could not create the Glue resource link: {e}" + + ok, out = _run_sql([ + f"""CREATE EXTERNAL SCHEMA {SCHEMA} + FROM DATA CATALOG DATABASE '{RESOURCE_LINK}' + IAM_ROLE '{ROLE_ARN}' + REGION '{REGION}' CATALOG_ID '{account}'""" + ]) + if not ok: + return False, f"could not create external schema: {_error_reason(out)}" + print(f"[setup] created external schema {SCHEMA} over {RESOURCE_LINK}") + _setup_fixture_schema(account, fed_catalog_id) + return True, "ready" + + +def _setup_fixture_schema(account: str = "", fed_catalog_id: str = "") -> None: + """Mount the Spark-created fixtures, if any were provided. + + Deliberately best-effort: fixtures are an enrichment, not a prerequisite. If + they are missing the affected tests skip and say so, which is far better than + failing the whole run and losing the 60-odd checks that need nothing from EMR. + """ + global FIXTURE_DB + if not FIXTURE_DB: + print("[setup] no REDSHIFT_FIXTURE_DB; features Redshift cannot create " + "will report read support as unmeasured") + return + + if STORAGE_MODE == "s3tables": + # The fixtures are an S3 Tables namespace, so they need their own resource + # link for the same reason the main schema does. + try: + _client("glue").create_database( + CatalogId=account, + DatabaseInput={ + "Name": FIXTURE_LINK, + "TargetDatabase": {"CatalogId": fed_catalog_id, + "DatabaseName": FIXTURE_DB}, + }, + ) + print(f"[setup] created fixture resource link {FIXTURE_LINK}") + except Exception as e: # noqa: BLE001 + if "AlreadyExists" not in type(e).__name__: + print(f"[setup] fixture resource link failed: {e}") + FIXTURE_DB = "" + return + ok, out = _run_sql([ + f"""CREATE EXTERNAL SCHEMA {FIXTURE_SCHEMA} + FROM DATA CATALOG DATABASE '{FIXTURE_LINK}' + IAM_ROLE '{ROLE_ARN}' + REGION '{REGION}' CATALOG_ID '{account}'""" + ]) + else: + ok, out = _run_sql([ + f"""CREATE EXTERNAL SCHEMA {FIXTURE_SCHEMA} + FROM DATA CATALOG DATABASE '{FIXTURE_DB}' + IAM_ROLE '{ROLE_ARN}'""" + ]) + if not ok: + print(f"[setup] could not mount fixtures from {FIXTURE_DB}: " + f"{_error_reason(out, 150)}") + FIXTURE_DB = "" + return + ok, out = _run_sql([ + f"SELECT tablename FROM svv_external_tables " + f"WHERE schemaname='{FIXTURE_SCHEMA}' ORDER BY tablename" + ]) + found = [l.strip() for l in out.splitlines() if l.strip()] if ok else [] + print(f"[setup] mounted {len(found)} fixtures as {FIXTURE_SCHEMA}: " + f"{', '.join(found) if found else 'none'}") + + +def teardown_catalog() -> None: + """Drop everything this run created, in dependency order.""" + ok, out = _run_sql([f"SELECT tablename FROM svv_external_tables WHERE schemaname='{SCHEMA}'"]) + if ok and out: + for name in [l.strip() for l in out.splitlines() if l.strip()]: + _run_sql([f"DROP TABLE {SCHEMA}.{name}"]) + _run_sql([f"DROP SCHEMA IF EXISTS {SCHEMA}"]) + + # Only the mount goes: the fixture tables are shared between the two storage + # mode runs and are owned by the EMR job, not by this one. + if FIXTURE_DB: + _run_sql([f"DROP SCHEMA IF EXISTS {FIXTURE_SCHEMA}"]) + if STORAGE_MODE == "s3tables" and TABLE_BUCKET_ARN: + try: + _client("glue").delete_database( + CatalogId=TABLE_BUCKET_ARN.split(":")[4], Name=FIXTURE_LINK) + print(f"[teardown] deleted fixture resource link {FIXTURE_LINK}") + except Exception as e: # noqa: BLE001 + print(f"[teardown] fixture link cleanup: {type(e).__name__}") + + if STORAGE_MODE == "s3tables": + # DROP TABLE removes the table from the bucket here, unlike the + # s3buckets mode, so only the namespace and the link are left. Any table + # a test created outside Redshift still needs sweeping first, since a + # namespace with tables in it cannot be deleted. + account = TABLE_BUCKET_ARN.split(":")[4] if TABLE_BUCKET_ARN else "" + try: + s3t = _client("s3tables") + listed = s3t.list_tables( + tableBucketARN=TABLE_BUCKET_ARN, namespace=S3T_NAMESPACE + ).get("tables", []) + for t in listed: + s3t.delete_table( + tableBucketARN=TABLE_BUCKET_ARN, + namespace=S3T_NAMESPACE, + name=t["name"], + ) + if listed: + print(f"[teardown] deleted {len(listed)} S3 Tables tables") + except Exception as e: # noqa: BLE001 + print(f"[teardown] S3 Tables table cleanup: {type(e).__name__}: {e}") + try: + _client("glue").delete_database(CatalogId=account, Name=RESOURCE_LINK) + print(f"[teardown] deleted Glue resource link {RESOURCE_LINK}") + except Exception as e: # noqa: BLE001 + print(f"[teardown] resource link cleanup: {type(e).__name__}: {e}") + try: + _client("s3tables").delete_namespace( + tableBucketARN=TABLE_BUCKET_ARN, namespace=S3T_NAMESPACE + ) + print(f"[teardown] deleted S3 Tables namespace {S3T_NAMESPACE}") + except Exception as e: # noqa: BLE001 + print(f"[teardown] namespace cleanup: {type(e).__name__}: {e}") + return + + try: + glue = _client("glue") + for t in glue.get_tables(DatabaseName=GLUE_DB).get("TableList", []): + glue.delete_table(DatabaseName=GLUE_DB, Name=t["Name"]) + glue.delete_database(Name=GLUE_DB) + print(f"[teardown] dropped Glue database {GLUE_DB}") + except Exception as e: # noqa: BLE001 + print(f"[teardown] Glue cleanup: {type(e).__name__}: {e}") + + # DROP TABLE only removes the catalog entry. Redshift leaves the Parquet and + # metadata objects in place, so without this every run would leak its entire + # data footprint into the bucket and pay for it indefinitely. + if not DATA_BUCKET: + return + prefix = f"redshift/{RUN_TAG}/" + try: + s3 = _client("s3") + removed = 0 + token = None + while True: + kwargs = {"Bucket": DATA_BUCKET, "Prefix": prefix, "MaxKeys": 1000} + if token: + kwargs["ContinuationToken"] = token + page = s3.list_objects_v2(**kwargs) + batch = [{"Key": o["Key"]} for o in page.get("Contents", [])] + if batch: + s3.delete_objects(Bucket=DATA_BUCKET, Delete={"Objects": batch}) + removed += len(batch) + token = page.get("NextContinuationToken") + if not token: + break + print(f"[teardown] deleted {removed} objects under s3://{DATA_BUCKET}/{prefix}") + except Exception as e: # noqa: BLE001 + print(f"[teardown] S3 cleanup: {type(e).__name__}: {e}") + + +def detect_version() -> str: + ok, out = _run_sql(["SELECT version()"]) + if ok and out: + m = re.search(r"Redshift ([\d.]+)", out) + return m.group(1) if m else out.strip()[:60] + return "unknown" + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def main(): + global REDSHIFT_VERSION + + print("=" * 70) + print(" Redshift Iceberg Feature Test Suite") + print("=" * 70) + print(f"Workgroup: {WORKGROUP}") + print(f"Database: {DATABASE}") + print(f"Auth: {'admin secret' if SECRET_ARN else 'caller IAM identity'}") + print(f"Storage mode: {STORAGE_MODE}") + print(f"Data bucket: {DATA_BUCKET or '(unset)'}") + print(f"Matrix platform: {MATRIX_PLATFORM_ID} ({MATRIX_DATA_PATH})") + if PLATFORM_LABEL: + print(f"Platform: {PLATFORM_LABEL}") + print(f"Versions: {', '.join(VERSIONS)}") + print(f"Run tag: {RUN_TAG}") + print() + + if not WORKGROUP: + print("[FATAL] REDSHIFT_WORKGROUP is not set") + sys.exit(1) + if not ROLE_ARN: + print("[FATAL] REDSHIFT_ROLE_ARN is required: the external schema must name " + "a role, because the auto-mounted catalog cannot write") + sys.exit(1) + if STORAGE_MODE == "s3buckets" and not DATA_BUCKET: + print("[FATAL] AWS_DATA_BUCKET is required for s3buckets") + sys.exit(1) + if STORAGE_MODE == "s3tables" and not TABLE_BUCKET_ARN: + print("[FATAL] AWS_TABLE_BUCKET_ARN is required for s3tables") + sys.exit(1) + + REDSHIFT_VERSION = detect_version() + print(f"Redshift version: {REDSHIFT_VERSION}") + + ok, message = setup_catalog() + if not ok: + print(f"[FATAL] {message}") + # Still emit a report, so the run records why the mode could not run + # rather than leaving an empty artifact behind. + results = [] + for version in VERSIONS: + for fn in ALL_TESTS: + r = TestResult( + fn.__name__.replace("test_", "").replace("_", "-"), + fn.__name__, version, + ) + r.result = "error" + r.details = f"setup failed: {message}" + results.append(r) + _write_reports(generate_report(results)) + sys.exit(1) + + only = os.environ.get("REDSHIFT_ONLY", "").strip() + tests = ALL_TESTS + if only: + wanted = {t.strip() for t in only.split(",") if t.strip()} + tests = [t for t in ALL_TESTS if t.__name__.replace("test_", "") in wanted] + print(f"[INFO] REDSHIFT_ONLY set; running {len(tests)} test(s)\n") + + results = [] + try: + for version in VERSIONS: + print(f"\n{'=' * 70}\n Format version {version.upper()}\n{'=' * 70}") + for fn in tests: + print(f"\n--- {fn.__name__} [{version}] ---") + try: + result = fn(version) + except Exception as e: # noqa: BLE001 + result = TestResult( + fn.__name__.replace("test_", "").replace("_", "-"), + fn.__name__, version, + ) + result.result = "error" + result.details = f"Unhandled exception: {e}" + results.append(result) + print(f" {result.result}: {result.details[:160]}") + finally: + teardown_catalog() + + report = generate_report(results) + _write_reports(report) + + s = report["summary"] + print(f"\n{'=' * 70}") + print(f" {s['passed']} passed, {s.get('partial', 0)} partial, " + f"{s['failed']} failed, {s['skipped']} skipped, " + f"{s['errors']} errors, {s['discrepancies']} discrepancies, " + f"{s['unverified']} unverified, " + f"{s.get('uncovered_features', 0)} uncovered matrix features") + print(f"{'=' * 70}") + + sys.exit(1 if (s["discrepancies"] > 0 or s["errors"] > 0 + or s.get("uncovered_features", 0) > 0) else 0) + + +def _write_reports(report: dict) -> None: + os.makedirs(REPORT_DIR, exist_ok=True) + stem = f"redshift-iceberg-test-report-{STORAGE_MODE}" + json_path = os.path.join(REPORT_DIR, f"{stem}.json") + with open(json_path, "w") as f: + json.dump(report, f, indent=2) + md = generate_markdown(report) + md_path = os.path.join(REPORT_DIR, f"{stem}.md") + with open(md_path, "w") as f: + f.write(md) + print(f"\nReports: {json_path}\n {md_path}") + print("\n" + md) + summary_file = os.environ.get("GITHUB_STEP_SUMMARY") + if summary_file: + with open(summary_file, "a") as f: + f.write(md) + + +if __name__ == "__main__": + main()