mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-20 06:07:05 +00:00
Chris LuandGitHub
fc75f16c30
* test(s3tables): expand Dremio Iceberg catalog test coverage
Restructure TestDremioIcebergCatalog into subtests and add three new
checks that go beyond a connectivity smoke test:
- ColumnProjection: SELECT id, label proves Dremio parsed the schema
served by the SeaweedFS REST catalog (the previous SELECT COUNT(*)
passed without exercising any column metadata).
- InformationSchemaColumns: verifies the table's columns are listed in
Dremio's INFORMATION_SCHEMA.COLUMNS in the expected ordinal order.
- InformationSchemaTables: verifies the table is registered in
INFORMATION_SCHEMA.TABLES.
All subtests share a single Dremio container startup, so total
runtime is unchanged.
* test(s3tables): exercise multi-level Iceberg namespaces from Dremio
Seed a 2-level Iceberg namespace (and a table inside it) via the REST
catalog before bootstrapping Dremio, then add a MultiLevelNamespace
subtest that scans the nested table by its dot-separated reference.
This relies on isRecursiveAllowedNamespaces=true (already set in the
Dremio source config) to surface the nested levels as folders. A
regression in either the SeaweedFS namespace path encoding (#8959-style)
or Dremio's recursive-namespace discovery would surface here.
Adds two helpers to keep the existing single-level call sites unchanged:
- createIcebergNamespaceLevels: namespace creation with []string levels
- createIcebergTableInLevels: table creation with []string levels and
unit-separator (0x1F) URL encoding for the namespace path component
* test(s3tables): verify Dremio reads PyIceberg-written rows
The previous Dremio subtests only scanned empty tables, so they did not
exercise the data path - just the catalog/metadata path. Add a
PyIceberg-based writer that materializes parquet files plus a snapshot
on a separate table before Dremio bootstraps, and two new subtests:
- ReadWrittenDataCount: SELECT COUNT(*) returns 3.
- ReadWrittenDataValues: SELECT id, label ORDER BY id returns the three
written rows with the expected (id, label) pairs.
The writer runs in a small image (Dockerfile.writer) built locally on
demand. It pip-installs pyiceberg+pyarrow once and reuses the layer
cache on subsequent runs. The CI workflow pre-pulls python:3.11-slim
to keep cold runs predictable.
The writer authenticates via the OAuth2 client_credentials flow that
SeaweedFS already exposes at /v1/oauth/tokens, mirroring the Go-side
helper used for REST-API table creation.
* test(s3tables): fix Dremio writer required-field schema mismatch
PyIceberg's append() compatibility check rejects an arrow column whose
nullability does not match the Iceberg field. The table schema declares
id as `required long`, but the default pyarrow int64 column is nullable
- so the writer failed with:
1: id: required long vs. 1: id: optional long
Declare an explicit pyarrow schema with nullable=False on id and
nullable=True on label to match the Iceberg side.
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Append rows to an existing Iceberg table via the SeaweedFS REST catalog.
|
|
|
|
Used by the Dremio integration test to materialize data files so a downstream
|
|
SELECT from Dremio verifies the read path against non-empty results.
|
|
|
|
Usage:
|
|
python3 append_rows.py \\
|
|
--catalog-url http://localhost:8181 \\
|
|
--warehouse s3://my-bucket \\
|
|
--prefix my-bucket \\
|
|
--s3-endpoint http://localhost:8333 \\
|
|
--access-key AKIA... --secret-key wJalr... \\
|
|
--namespace foo --namespace bar \\
|
|
--table events
|
|
"""
|
|
|
|
import argparse
|
|
import sys
|
|
|
|
import pyarrow as pa
|
|
from pyiceberg.catalog import load_catalog
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument("--catalog-url", required=True)
|
|
p.add_argument("--warehouse", required=True, help="s3://<bucket-name>")
|
|
p.add_argument("--prefix", required=True, help="REST catalog prefix (table bucket name)")
|
|
p.add_argument("--s3-endpoint", required=True, help="http://host:port")
|
|
p.add_argument("--access-key", required=True)
|
|
p.add_argument("--secret-key", required=True)
|
|
p.add_argument("--region", default="us-east-1")
|
|
p.add_argument(
|
|
"--namespace",
|
|
action="append",
|
|
required=True,
|
|
help="One per level (e.g. --namespace foo --namespace bar for foo.bar).",
|
|
)
|
|
p.add_argument("--table", required=True)
|
|
args = p.parse_args()
|
|
|
|
# `credential` triggers OAuth2 client_credentials against
|
|
# <catalog_uri>/v1/oauth/tokens, matching the helper Go test uses for
|
|
# REST-API table creation. The s3.* keys are needed for parquet writes.
|
|
catalog = load_catalog(
|
|
"rest",
|
|
**{
|
|
"type": "rest",
|
|
"uri": args.catalog_url,
|
|
"warehouse": args.warehouse,
|
|
"prefix": args.prefix,
|
|
"credential": f"{args.access_key}:{args.secret_key}",
|
|
"s3.access-key-id": args.access_key,
|
|
"s3.secret-access-key": args.secret_key,
|
|
"s3.endpoint": args.s3_endpoint,
|
|
"s3.region": args.region,
|
|
"s3.path-style-access": "true",
|
|
},
|
|
)
|
|
|
|
table_id = tuple(args.namespace) + (args.table,)
|
|
table = catalog.load_table(table_id)
|
|
|
|
# Match the Iceberg table schema: id is `required long`, label is
|
|
# `optional string`. Default pyarrow columns are nullable, which fails
|
|
# PyIceberg's required-field compatibility check.
|
|
arrow_schema = pa.schema(
|
|
[
|
|
pa.field("id", pa.int64(), nullable=False),
|
|
pa.field("label", pa.string(), nullable=True),
|
|
]
|
|
)
|
|
arrow_table = pa.Table.from_pydict(
|
|
{
|
|
"id": [1, 2, 3],
|
|
"label": ["one", "two", "three"],
|
|
},
|
|
schema=arrow_schema,
|
|
)
|
|
table.append(arrow_table)
|
|
print(f"appended {arrow_table.num_rows} rows to {'.'.join(table_id)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|