feat: first attempt at item table logic

.github/workflows/ci.yml

@@ -0,0 +1,41 @@
+name: Test
+
+on:
+  workflow_dispatch:
+  push:
+    branches:
+    - main
+    - develop
+    tags:
+  pull_request:
+
+jobs:
+  tests:
+    strategy:
+      matrix:
+        python-version: [3.11, 3.12, 3.13, 3.14]
+
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v5
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v7
+        with:
+          version: "0.10.*" 
+          python-version: ${{ matrix.python-version }}
+          enable-cache: true
+
+      - name: Run pre-commit
+        run: uv run pre-commit run --all-files
+
+      - name: Run tests
+        run: uv run pytest --cov-branch --cov-report=xml
+
+      - name: Upload coverage to Codecov
+        if: matrix.python-version == '3.14'
+        uses: codecov/codecov-action@v5
+        with:
+          token: ${{ secrets.CODECOV_TOKEN }}
+          files: coverage.xml

.gitignore

@@ -0,0 +1,20 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+.coverage
+coverage.xml
+
+# Virtual environments
+.venv
+
+minio-data/
+
+# dev docs
+dev-docs/
+
+data/
+.worktrees/

.pre-commit-config.yaml

@@ -0,0 +1,17 @@
+default_stages: [pre-commit, pre-push]
+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.15.2
+    hooks:
+    - id: ruff
+      args: ["--fix", "--show-fixes"]
+    - id: ruff-format
+
+  - repo: local
+    hooks:
+      - id: ty
+        name: ty (type check)
+        entry: uv run ty check
+        language: system
+        types: [python]
+        pass_filenames: false

.pyiceberg.yaml

@@ -0,0 +1,9 @@
+catalog:
+  default:
+    type: rest
+    uri: http://localhost:8181
+    warehouse: s3://warehouse/
+    s3.endpoint: http://localhost:9000
+    s3.access-key-id: admin
+    s3.secret-access-key: password
+    s3.path-style-access: "true"

.python-version

@@ -0,0 +1 @@
+3.13

README.md

@@ -1 +1,191 @@
 # icestac
+
+icestac maps STAC concepts onto Apache Iceberg.
+
+The current implementation covers STAC Items with schema utilities for the flattened item layout defined by [STAC GeoParquet](https://radiantearth.github.io/stac-geoparquet-spec/), a small write API, and runnable PyIceberg examples.
+
+Each collection gets one Iceberg table named with its STAC collection ID. Clients can locate an item table from a collection ID without a metadata lookup. Each row represents one item, and an item's identity is scoped to its collection. Requesting an item by collection and ID should return one current record, not multiple versions.
+
+The `collections` table name is reserved for planned collection metadata. Collection metadata is not implemented yet.
+
+## Storage conventions
+
+Use PyIceberg for catalog and table administration, including namespace and table creation, partitioning, sorting, and table properties. icestac does not choose a default table layout.
+
+The schema utilities and `put_items` handle the STAC-facing parts. They work with an Arrow table (`pyarrow.Table` or `arro3.core.Table`) using the [STAC GeoParquet](https://radiantearth.github.io/stac-geoparquet-spec/) column mapping: item properties are top-level columns, while geometry is WKB and links and assets remain nested structures. This describes the in-memory Arrow input, not a compliant STAC GeoParquet file; file compliance also requires GeoParquet and STAC metadata. `get_schema_from_items` checks the required fields and their basic Arrow types, then creates an Iceberg schema. It does not perform full STAC validation.
+
+The inferred schema marks `id` as an Iceberg identifier field. Iceberg identifier fields are schema metadata, not uniqueness constraints. Starting with a table that has unique `(collection, id)` pairs and writing through `put_items` preserves the one-current-row convention. Native writes that bypass `put_items`, such as appends, can create duplicate IDs and violate it.
+
+## Put items
+
+Create a table for one collection with native PyIceberg, then pass its items to `put_items`:
+
+```python
+from pyiceberg.catalog import load_catalog
+
+from icestac.constants import DEFAULT_NAMESPACE
+from icestac.schema import get_schema_from_items
+from icestac.write import put_items
+
+# items is an Arrow table containing items from one collection.
+collection_id = "my-collection"
+catalog = load_catalog()
+catalog.create_namespace_if_not_exists(DEFAULT_NAMESPACE)
+schema = get_schema_from_items(items)
+table = catalog.create_table(
+    identifier=(DEFAULT_NAMESPACE, collection_id),
+    schema=schema,
+)
+put_items(table, items)
+```
+
+`put_items` accepts items from one collection and checks that their `collection` values match the collection named by the table. It rejects duplicate IDs within an input batch. Matching items are complete replacements, so omitted optional fields are cleared, including nested link fields. New IDs are inserted.
+
+Collection IDs can contain periods. Pass catalog identifiers as tuples such as `(namespace, collection_id)` instead of interpolated dotted strings; a catalog backend may still reject a particular ID. Do not use `collections` as an item table name.
+
+For append semantics, use native PyIceberg:
+
+```python
+table.append(df=items)
+```
+
+### Schema evolution
+
+New fields raise an error unless you pass `evolve_schema=True`. Schema evolution and the replacement write happen in one Iceberg transaction:
+
+```python
+put_items(table, items, evolve_schema=True)
+```
+
+To update a schema without writing items, use native PyIceberg:
+
+```python
+from pyiceberg.catalog import load_catalog
+
+from icestac.constants import DEFAULT_NAMESPACE
+from icestac.schema import get_schema_from_items
+
+catalog = load_catalog()
+table = catalog.load_table((DEFAULT_NAMESPACE, collection_id))
+with table.update_schema() as update:
+    update.union_by_name(get_schema_from_items(items))
+```
+
+### Concurrent writes
+
+Each batch, including optional schema evolution, commits as one Iceberg transaction. A concurrent commit conflict raises `CommitFailedException`; refresh the table and retry from current state rather than replaying a stale replacement. A `CommitStateUnknownException` means the commit outcome is unknown, so refresh and reconcile before retrying.
+
+## Table layout
+
+Choose the layout when creating the table. For example, partition by month of `datetime`, sort by `datetime`, and set a Parquet row group limit:
+
+```python
+from pyiceberg.partitioning import PartitionField, PartitionSpec
+from pyiceberg.table import TableProperties
+from pyiceberg.table.sorting import SortField, SortOrder
+from pyiceberg.transforms import MonthTransform
+
+schema = get_schema_from_items(items)
+datetime_id = schema.find_field("datetime").field_id
+table = catalog.create_table(
+    identifier=(DEFAULT_NAMESPACE, collection_id),
+    schema=schema,
+    partition_spec=PartitionSpec(
+        PartitionField(
+            source_id=datetime_id,
+            field_id=1000,
+            transform=MonthTransform(),
+            name="datetime_month",
+        )
+    ),
+    sort_order=SortOrder(SortField(source_id=datetime_id)),
+    properties={TableProperties.PARQUET_ROW_GROUP_LIMIT: "50000"},
+)
+```
+
+Omit `partition_spec` for an unpartitioned table. Interval items with a null `datetime` go into a null partition. To partition them by their start time, use the `start_datetime` field with `MonthTransform()`.
+
+Geometry uses WKB in an Iceberg binary column. The output files lack the metadata required for GeoParquet and STAC GeoParquet compliance.
+
+## Local demo
+
+Run these commands from the repository root. You need [uv](https://docs.astral.sh/uv/), Docker Compose, and DuckDB for the query examples.
+
+### Start the services
+
+```bash
+docker compose up -d
+```
+
+- Iceberg REST catalog: `http://localhost:8181`
+- MinIO API: `http://localhost:9000`; console: `http://localhost:9001`
+- The `mc` service creates the `warehouse` bucket.
+
+The checked-in `.pyiceberg.yaml` points to these services. The credentials (`admin` / `password`) are for local development only. See [PyIceberg configuration](https://py.iceberg.apache.org/configuration/) for other environments.
+
+### Load HLS items
+
+```bash
+uv run main.py
+```
+
+The demo downloads January–August 2026 HLS STAC GeoParquet from public S3 into `data/`, reuses cached files, and writes each month's items to the `HLSS30_2.0` collection table. Each batch must fit in memory.
+
+It keeps the original `HLSS30_2.0` collection ID, partitions by month of `datetime`, sorts by DuckDB's Hilbert index of each bbox's lower-left coordinates, and caps Parquet row groups at 50,000 rows. Compatible schema changes use `evolve_schema=True`; rerunning the demo reuses the existing table and replaces the cached batches again.
+
+### Query with DuckDB
+
+The query selects items overlapping the contiguous United States bounding box (longitude -125 to -66, latitude 24 to 50) from April 1, 2026 inclusive through July 1, 2026 exclusive; the bbox predicates test bounding-box overlap, not exact geometry intersection.
+
+```sql
+INSTALL iceberg; LOAD iceberg;
+INSTALL httpfs; LOAD httpfs;
+
+CREATE OR REPLACE SECRET minio (
+    TYPE S3,
+    KEY_ID 'admin',
+    SECRET 'password',
+    ENDPOINT 'localhost:9000',
+    USE_SSL false,
+    URL_STYLE 'path'
+);
+
+ATTACH 'icestac' AS catalog (
+    TYPE ICEBERG,
+    ENDPOINT 'http://localhost:8181',
+    AUTHORIZATION_TYPE 'none'
+);
+
+SELECT count(*) FROM catalog.icestac."HLSS30_2.0";
+
+EXPLAIN ANALYZE SELECT count(*)
+FROM catalog.icestac."HLSS30_2.0"
+WHERE bbox.xmin <= -90.0
+  AND bbox.xmax >= -100.0
+  AND bbox.ymin <= 50.0
+  AND bbox.ymax >= 40.0
+  AND datetime >= TIMESTAMPTZ '2026-04-01 00:00:00+00'
+  AND datetime < TIMESTAMPTZ '2026-07-01 00:00:00+00'
+;
+
+```
+
+### Delete the demo table
+
+**This deletes the table and its data.**
+
+```python
+from pyiceberg.catalog import load_catalog
+
+from icestac.constants import DEFAULT_NAMESPACE
+
+load_catalog().purge_table((DEFAULT_NAMESPACE, "HLSS30_2.0"))
+```
+
+To remove only the catalog entry and keep the files, use `drop_table(...)` instead.
+
+## Tests
+
+```bash
+uv run pytest
+```

docker-compose.yml

@@ -0,0 +1,52 @@
+services:
+  rest:
+    image: apache/iceberg-rest-fixture
+    container_name: iceberg-rest
+    networks:
+      iceberg_net:
+    ports:
+      - 8181:8181
+    environment:
+      - AWS_ACCESS_KEY_ID=admin
+      - AWS_SECRET_ACCESS_KEY=password
+      - AWS_REGION=us-east-1
+      - CATALOG_WAREHOUSE=s3://warehouse/
+      - CATALOG_IO__IMPL=org.apache.iceberg.aws.s3.S3FileIO
+      - CATALOG_S3_ENDPOINT=http://minio:9000
+  minio:
+    image: minio/minio
+    container_name: minio
+    environment:
+      - MINIO_ROOT_USER=admin
+      - MINIO_ROOT_PASSWORD=password
+      - MINIO_DOMAIN=minio
+    networks:
+      iceberg_net:
+        aliases:
+          - warehouse.minio
+    ports:
+      - 9001:9001
+      - 9000:9000
+    volumes:
+      - ./minio-data:/data
+    command: ["server", "/data", "--console-address", ":9001"]
+  mc:
+    depends_on:
+      - minio
+    image: minio/mc
+    container_name: mc
+    networks:
+      iceberg_net:
+    environment:
+      - AWS_ACCESS_KEY_ID=admin
+      - AWS_SECRET_ACCESS_KEY=password
+      - AWS_REGION=us-east-1
+    entrypoint: |
+      /bin/sh -c "
+      until (/usr/bin/mc alias set minio http://minio:9000 admin password) do echo '...waiting...' && sleep 1; done;
+      /usr/bin/mc mb -p minio/warehouse;
+      /usr/bin/mc policy set public minio/warehouse;
+      tail -f /dev/null
+      "
+networks:
+  iceberg_net: