Project Nessie

# Nessie 0.109.0 and MinIO (as S3) in Docker, on one network:
#   docker network create lake
#   docker run -d --name minio --network lake -p 19000:9000 minio/minio server /data
#   docker exec minio sh -c 'mc alias set local http://localhost:9000 minioadmin minioadmin && mc mb local/lake'
#   docker run -d --name nessie --network lake -p 19120:19120 \
#     -e nessie.catalog.default-warehouse=lake \
#     -e nessie.catalog.warehouses.lake.location=s3://lake/ \
#     -e nessie.catalog.service.s3.default-options.endpoint=http://minio:9000/ \
#     -e nessie.catalog.service.s3.default-options.external-endpoint=http://localhost:19000/ \
#     -e nessie.catalog.service.s3.default-options.path-style-access=true \
#     -e nessie.catalog.service.s3.default-options.region=us-east-1 \
#     -e nessie.catalog.service.s3.default-options.access-key=urn:nessie-secret:quarkus:minio \
#     -e minio.name=minioadmin -e minio.secret=minioadmin \
#     ghcr.io/projectnessie/nessie:0.109.0
# pip install "pyiceberg[pyarrow,s3fs]"   (pyiceberg 0.12)

import json, re, urllib.request
import pyarrow as pa
from pyiceberg.catalog.rest import RestCatalog

NESSIE = "http://localhost:19120"

def api(method, path, body=None):  # Nessie's own REST API (v2)
    req = urllib.request.Request(NESSIE + "/api/v2" + path, method=method,
                                 data=json.dumps(body).encode() if body else None,
                                 headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(req) as r:
        return json.load(r)

def catalog(branch):  # Iceberg REST: the branch goes at the end of the URI
    return RestCatalog(branch, uri=f"{NESSIE}/iceberg/{branch}", **{
        # ask Nessie to sign each S3 request instead of handing out keys
        "header.X-Iceberg-Access-Delegation": "remote-signing"})

# 1. A namespace on main
main = catalog("main")
main.create_namespace("sales")

# 2. Branch the whole catalog: one pointer, no data copied
head = api("GET", "/trees/main")["reference"]
dev = api("POST", "/trees?name=dev&type=BRANCH", head)["reference"]
print(dev["name"], dev["hash"] == head["hash"])         # dev True

# 3. Create and fill a table on dev only
devcat = catalog("dev")
orders = devcat.create_table("sales.orders",
                             pa.schema([("id", pa.int64()), ("amount", pa.float64())]))
orders.append(pa.table({"id": [1, 2, 3], "amount": [9.5, 20.0, 4.25]}))

print("s3.access-key-id" in orders.io.properties)     # False: Nessie signed each request
print(devcat.list_tables("sales"))                      # [('sales', 'orders')]
print(main.list_tables("sales"))                        # []: main is unchanged

# 4. Merge dev into main: every change on dev lands in one atomic step
main_ref = api("GET", "/trees/main")["reference"]
dev_ref = api("GET", "/trees/dev")["reference"]
merge = api("POST", f"/trees/main@{main_ref['hash']}/history/merge",
            {"fromRefName": "dev", "fromHash": dev_ref["hash"]})
print(merge["wasSuccessful"])                           # True

rows = main.load_table("sales.orders").scan().to_arrow()
print(rows.num_rows, sum(rows["amount"].to_pylist()))   # 3 33.75

def log(ref):  # newest first, hashes cut to 8 characters
    return [re.sub(r"([0-9a-f]{8})[0-9a-f]{56}", r"\1", e["commitMeta"]["message"])
            for e in api("GET", f"/trees/{ref}/history")["logEntries"]]

print(log("dev"))
print(log("main"))
# ['Update ICEBERG_TABLE sales.orders', 'Create ICEBERG_TABLE sales.orders', 'update namespace sales']
# ['Merged dev at 1b6cc363 into main at 6cce0539', 'update namespace sales']