From b2a0664e60d40dbf45869cfdffa39dfe41c6b31a Mon Sep 17 00:00:00 2001 From: Angel Conde Date: Fri, 31 Jul 2026 09:21:50 +0200 Subject: [PATCH 1/3] feat(flink): Restore report parity on the merged Flink suite PR #32 replaced the Flink suite with a stronger, Docker-driven one, but its report dropped the fields the EMR, Glue, Dataproc and Spark reports carry, so a Flink run could no longer be read the same way as the others. In particular it had no notion of matrix coverage, which is what stops a feature added to features.json from sitting untested indefinitely. Add platform, platform_label, catalog_mode, versions_tested and coverage to the JSON report plus summary.uncovered_features, render the matching header lines, the uncovered-features row and the coverage section in the markdown, and let uncovered features fail the run as they already do for Spark. Also make the comparison target configurable through MATRIX_PLATFORM_ID, MATRIX_DATA_PATH, PLATFORM_LABEL and MATRIX_CATALOG_MODE, so the suite is no longer pinned to the OSS Flink cells. Nothing about how tests execute changed: the docker/local modes, the 35 tests, their per-version runs and every result they produce are untouched. The two things the merged suite added that the older report lacked, the unverified count and the execution mode, are kept and now sit alongside the coverage row. Verified by building one synthetic result per test and version, then asserting the report fields and every markdown fragment, including re-rendering with a feature dropped to prove the uncovered section and the failing exit code work. Coverage reads 35/35 with no ids outside the matrix. --- .github/workflows/flink-tests.yml | 1 + tests/flink_feature_tests.py | 109 ++++++++++++++++++++++++++++-- 2 files changed, 104 insertions(+), 6 deletions(-) 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/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__": From 8348a285070ceee1da8b9786f082862156b2bb96 Mon Sep 17 00:00:00 2001 From: Angel Conde Date: Mon, 3 Aug 2026 10:47:22 +0200 Subject: [PATCH 2/3] feat(aws): Run the feature suite on Redshift Serverless, both modes Adds Redshift to the managed-engine drivers. A low-RPU Serverless workgroup joins the existing stack, and the suite is driven through the Redshift Data API rather than shipped to a cluster, so it runs in the runner and needs no bundle or entry point. Two things about Redshift shaped the design. Writes need an external schema that names the 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. S3 Tables needs a Glue resource link. A table bucket is a federated Glue catalog and Redshift cannot name one directly; putting the federated path in CATALOG_ID silently resolves against the default catalog instead, so the schema is created and every SELECT then raises EntityNotFoundException. With a resource link in the default catalog, and CATALOG_ID set to the plain account id, Redshift creates, reads, updates and deletes Iceberg tables in S3 Tables. Some features cannot be tested against a table Redshift is able to create: it refuses format-version 3 outright. Spark on EMR builds those fixtures instead and the suite reads and writes them, which separates "cannot write" from "cannot read". Redshift turns out to read v3 tables and apply deletion vectors correctly while refusing every v3 write, so those cells are partial rather than absent. A new "partial" result records that, and compute_match requires partial to meet partial exactly, so the level is falsifiable instead of matching anything. Also fixes two leaks found on the way. DROP TABLE removes only the Glue entry in the s3buckets mode, leaving the data behind, so the suite deletes its own S3 prefix; and the shared teardown never swept the fixture warehouse, which holds real Parquet. Verified against Redshift 1.0.365190 on 8 RPU, both modes, with the fixtures in place and no leftover billable resources afterwards. --- .github/workflows/aws-platform-tests.yml | 45 +- infra/aws/README.MD | 82 + infra/aws/aws.yaml | 245 ++- tests/aws/platform_common.py | 13 +- tests/aws/redshift_fixtures.py | 239 +++ tests/aws/run_redshift.py | 199 ++ tests/aws/run_redshift_fixtures.py | 158 ++ tests/aws/teardown.py | 17 +- tests/redshift_feature_tests.py | 2243 ++++++++++++++++++++++ 9 files changed, 3230 insertions(+), 11 deletions(-) create mode 100644 tests/aws/redshift_fixtures.py create mode 100644 tests/aws/run_redshift.py create mode 100644 tests/aws/run_redshift_fixtures.py create mode 100644 tests/redshift_feature_tests.py 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/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/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/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() From 352c02ee0d3e5be87f0bbc0c4cc971885c0b24db Mon Sep 17 00:00:00 2001 From: Angel Conde Date: Mon, 3 Aug 2026 10:49:40 +0200 Subject: [PATCH 3/3] fix(data): Correct the Redshift cells from the Serverless measurements Fourteen cells disagreed with what the engine actually does. Both modes claimed copy-on-write and time travel; neither exists. Redshift is merge-on-read only and refuses the write.delete.mode property outright, and no time-travel syntax is accepted at all, with no snapshot metadata table to fall back on. The S3 Tables file was the staler of the two and had the catalog story backwards. It said S3 Tables uses its own built-in catalog rather than Glue, and that Redshift consumes the REST API. It is the other way round: Glue is the only way Redshift reaches S3 Tables, through a resource link, and it has no Iceberg REST client whatsoever. Three cells become partial because the feature is genuinely half-present. Redshift reads format-version 3 tables and applies deletion vectors correctly, but refuses every v3 write; and on S3 Tables, PARTITIONED BY at CREATE is accepted and then silently discarded while ALTER ADD PARTITION FIELD does apply, so transform partitioning takes two statements instead of one. Each entry now carries the engine's own error text in its caveats, so a future reader can tell why a cell says what it says without rerunning anything. equality-deletes is deliberately left alone. Its notes claim Redshift reads equality deletes, and all that could be established is that Redshift never writes them, which does not contradict a read claim. No fixture carrying equality deletes could be produced, because Spark writes position deletes, so the cell stays unverified rather than being changed on a hunch. Both storage modes now report zero discrepancies. --- .../s3buckets/redshift-s3/redshift-s3.json | 61 +++++++--- .../aws/s3tables/redshift-s3/redshift-s3.json | 109 +++++++++++++----- 2 files changed, 123 insertions(+), 47 deletions(-) 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" + ] } } }