|
| 1 | +--- |
| 2 | +apiVersion: v1 |
| 3 | +kind: ConfigMap |
| 4 | +metadata: |
| 5 | + name: delta-check-script |
| 6 | +data: |
| 7 | + check-delta.py: | |
| 8 | + #!/usr/bin/env python |
| 9 | + # |
| 10 | + # This script is used to run a series of SQL statements against the Trino cluster to verify that the Delta Lake connector is working correctly. |
| 11 | + # |
| 12 | + # It assumes that data is already present in the S3 location and that the Trino cluster is configured to access it. The script will: |
| 13 | + # 1. Register the Delta table using the system.register_table procedure. |
| 14 | + # 2. Run a SELECT query to verify that the expected data is present. |
| 15 | + # 3. Run a DELETE query to remove some data. |
| 16 | + # 4. Run another SELECT query to verify that the data was deleted. |
| 17 | + # |
| 18 | + import argparse |
| 19 | + import sys |
| 20 | +
|
| 21 | + import trino |
| 22 | +
|
| 23 | + if not sys.warnoptions: |
| 24 | + import warnings |
| 25 | +
|
| 26 | + warnings.simplefilter("ignore") |
| 27 | +
|
| 28 | + def get_connection(username, password, coordinator): |
| 29 | + conn = trino.dbapi.connect( |
| 30 | + host=coordinator, |
| 31 | + port=8443, |
| 32 | + user=username, |
| 33 | + http_scheme="https", |
| 34 | + auth=trino.auth.BasicAuthentication(username, password), |
| 35 | + session_properties={"query_max_execution_time": "60s"}, |
| 36 | + ) |
| 37 | + conn._http_session.verify = False |
| 38 | + return conn |
| 39 | +
|
| 40 | +
|
| 41 | + def run_query(connection, query): |
| 42 | + print(f"[DEBUG] Executing query {query}") |
| 43 | + cursor = connection.cursor() |
| 44 | + cursor.execute(query) |
| 45 | + return cursor.fetchall() |
| 46 | +
|
| 47 | +
|
| 48 | + if __name__ == "__main__": |
| 49 | + all_args = argparse.ArgumentParser() |
| 50 | + all_args.add_argument( |
| 51 | + "-c", |
| 52 | + "--coordinator", |
| 53 | + required=True, |
| 54 | + help="Trino Coordinator Host to connect to", |
| 55 | + ) |
| 56 | + args = vars(all_args.parse_args()) |
| 57 | +
|
| 58 | + coordinator = args["coordinator"] |
| 59 | +
|
| 60 | + print("[INFO] Running Delta SQL commands") |
| 61 | + connection = get_connection("admin", "admin", coordinator) |
| 62 | +
|
| 63 | + statement = "DROP TABLE IF EXISTS delta.default.delta_table" |
| 64 | + run_query(connection, statement) |
| 65 | +
|
| 66 | + # Must use the register_table procedure to create the table because data is already present in the S3 location |
| 67 | + # and Trino's CREATE TABLE AS doesn't support that. |
| 68 | + statement = "CALL delta.system.register_table(schema_name => 'default', table_name => 'delta_table', table_location => 's3a://trino/delta-table')" |
| 69 | + run_query(connection, statement) |
| 70 | +
|
| 71 | + statement = "select count(*) from delta.default.delta_table where text='text5'" |
| 72 | + result = run_query(connection, statement) |
| 73 | + assert result[0][0] == 5000, ( |
| 74 | + f"Expected 5000 rows, got {result}\nQuery: {statement}" |
| 75 | + ) |
| 76 | +
|
| 77 | + statement = "delete from delta.default.delta_table where text = 'text5'" |
| 78 | + run_query(connection, statement) |
| 79 | +
|
| 80 | + statement = "select count(*) from delta.default.delta_table where text='text5'" |
| 81 | + result = run_query(connection, statement) |
| 82 | + assert result[0][0] == 0, ( |
| 83 | + f"Expected 0 rows, got {result}\nQuery: {statement}" |
| 84 | + ) |
| 85 | +
|
| 86 | + print("[SUCCESS] Delta checks succeeded.") |
| 87 | +--- |
| 88 | +apiVersion: batch/v1 |
| 89 | +kind: Job |
| 90 | +metadata: |
| 91 | + name: delta-check |
| 92 | +spec: |
| 93 | + backoffLimit: 0 |
| 94 | + template: |
| 95 | + spec: |
| 96 | + serviceAccountName: integration-tests-sa |
| 97 | + restartPolicy: Never |
| 98 | + containers: |
| 99 | + - name: trino-test-helper |
| 100 | + image: oci.stackable.tech/sdp/testing-tools:0.2.0-stackable0.0.0-dev |
| 101 | + command: ["python", "/tmp/check-delta.py", "-c", "trino-coordinator"] |
| 102 | + resources: |
| 103 | + requests: |
| 104 | + cpu: "250m" |
| 105 | + memory: "64Mi" |
| 106 | + limits: |
| 107 | + cpu: "500m" |
| 108 | + memory: "64Mi" |
| 109 | + volumeMounts: |
| 110 | + - name: delta-check-script |
| 111 | + mountPath: /tmp/check-delta.py |
| 112 | + subPath: check-delta.py |
| 113 | + volumes: |
| 114 | + - name: delta-check-script |
| 115 | + configMap: |
| 116 | + name: delta-check-script |
0 commit comments