Feature/v1 metadata analysis

analysis/structured_metadata/v1.ipynb

@@ -2,38 +2,44 @@
  "cells": [
   {
    "cell_type": "code",
-   "execution_count": 9,
+   "execution_count": null,
    "id": "77751b75",
    "metadata": {},
    "outputs": [],
    "source": [
+    "import orjson\n",
+    "import genson\n",
+    "import natsort\n",
+    "import thefuzz.fuzz\n",
+    "import thefuzz.process\n",
+    "import rich\n",
     "from data_index.iceberg_config import S3TablesCatalogConfig, IcebergTableConfig"
    ]
   },
   {
    "cell_type": "code",
-   "execution_count": 8,
+   "execution_count": null,
    "id": "f40ca1d4",
    "metadata": {},
    "outputs": [],
    "source": [
-    "# --- Sink config ---\n",
+    "# --- Sinks config ---\n",
     "data_index_catalog_config = S3TablesCatalogConfig(\n",
     "    region=\"ap-southeast-2\",\n",
     "    arn=\"arn:aws:s3tables:ap-southeast-2:704910415367:bucket/data-index\",\n",
     ")\n",
     "\n",
-    "structured_metadata_table_config = IcebergTableConfig(\n",
+    "structured_table = IcebergTableConfig(\n",
     "    catalog_config=data_index_catalog_config,\n",
     "    namespace=\"data_index\",\n",
     "    table_name=f\"structured_metadata_v1\",\n",
-    ")\n",
+    ").load()\n",
     "\n",
-    "unstructured_metadata_table_config = IcebergTableConfig(\n",
+    "unstructured_table = IcebergTableConfig(\n",
     "    catalog_config=data_index_catalog_config,\n",
     "    namespace=\"data_index\",\n",
     "    table_name=\"unstructured_metadata\",\n",
-    ")"
+    ").load()"
    ]
   },
   {
@@ -43,7 +49,85 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "data_index_catalog_config."
+    "# Read into memory as polars.DataFrame\n",
+    "sdf = structured_table.scan().to_polars()\n",
+    "udf = unstructured_table.scan().to_polars()\n",
+    "display(sdf)\n",
+    "display(udf)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "f3cf4ee1",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Construct generator\n",
+    "metadata = (orjson.loads(metadata) for metadata in udf[\"metadata\"])\n",
+    "\n",
+    "# Build schema\n",
+    "schema_builder = genson.SchemaBuilder()\n",
+    "for metadata in metadata:\n",
+    "    schema_builder.add_object(metadata)\n",
+    "\n",
+    "# Extract schema\n",
+    "schema = schema_builder.to_schema()"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "897a4c30",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Sort global attribute keys naturally (human-like)\n",
+    "sorted_global_attribute_keys = natsort.natsorted(\n",
+    "    seq=[k for k in schema[\"properties\"][\"global_attrs\"][\"properties\"]],\n",
+    "    alg=natsort.IGNORECASE,\n",
+    ")\n",
+    "print(sorted_global_attribute_keys)"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "992de4fc",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "# Tuning parameters\n",
+    "SIMILARITY_THRESHOLD = 80\n",
+    "# Using token_sort_ratio handles underscores and word order swaps well\n",
+    "SCORER = thefuzz.fuzz.token_sort_ratio \n",
+    "\n",
+    "clusters = []\n",
+    "visited = set()\n",
+    "\n",
+    "for key in sorted_global_attribute_keys:\n",
+    "    if key in visited:\n",
+    "        continue\n",
+    "    \n",
+    "    # Find all items in the list that match the current key above our threshold\n",
+    "    matches = thefuzz.process.extractBests(\n",
+    "        query=key, \n",
+    "        choices=sorted_global_attribute_keys, \n",
+    "        scorer=SCORER, \n",
+    "        score_cutoff=SIMILARITY_THRESHOLD\n",
+    "    )\n",
+    "    \n",
+    "    # Filter matches to only include keys we haven't clustered yet\n",
+    "    current_cluster = [match[0] for match in matches if match[0] not in visited]\n",
+    "    \n",
+    "    if current_cluster:\n",
+    "        clusters.append(current_cluster)\n",
+    "        # Mark all keys in this cluster as visited so they aren't processed again\n",
+    "        visited.update(current_cluster)\n",
+    "\n",
+    "# Print the resulting list of lists\n",
+    "import pprint\n",
+    "pprint.pprint(clusters)"
    ]
   }
  ],

docs/managing-s3-tables.md

@@ -15,4 +15,21 @@ aws s3tables delete-table \
 aws s3tables delete-table-bucket \
 --region ap-southeast-2 \
 --table-bucket-arn <value>
-```
+```
+
+## IAM Permissions
+
+The default cluster flow reads from the live inventory S3 Table and writes to two data-index S3 Tables (with flow-run-specific suffixes during cluster runs).
+
+
+### Scopes
+| Access scope | Table bucket ARN | Namespace / table(s) | Required IAM actions |
+| --- | --- | --- | --- |
+| Read source inventory | `arn:aws:s3tables:ap-southeast-2:104044260116:bucket/aws-s3` | `b_imos-data.inventory` | `s3tables:GetTable`, `s3tables:GetTableMetadataLocation`, `s3tables:GetTableData` |
+| Provision + write structured/unstructured outputs | `arn:aws:s3tables:ap-southeast-2:704910415367:bucket/data-index` | Namespace: `data_index`; Tables: `structured_metadata_v*`, `unstructured_metadata*` (including flow-run suffix variants) | `s3tables:CreateNamespace`, `s3tables:CreateTable`, `s3tables:GetTable`, `s3tables:GetTableMetadataLocation`, `s3tables:UpdateTableMetadataLocation`, `s3tables:PutTableData`, `s3tables:GetTableData` |
+
+### Why these actions are needed
+
+- `LiveS3InventorySource` reads the inventory Iceberg table via S3 Tables catalog APIs.
+- `StructuredS3TableSink` and `UnstructuredS3TableSink` call `provision()` (create namespace/table if missing) and then append rows, which updates table metadata locations and writes table data.
+- `run_index_cluster()` appends `flow_run_id` to sink table names for test/isolation runs, so IAM should allow matching wildcard table names in the `data_index` namespace.

src/data_index/metadata_extractor/attribute_netcdf_extractor.py

@@ -57,6 +57,9 @@ def _extract_structured(
         :returns: Structured metadata row.
         """
 
+        # TODO:
+        # For dimensions, capture the sizes of the dimensions not just the names
+
         metadata_kwargs = {
             "s3_uri": s3_uri,
             "file_format": file_format,