diff --git a/.github/dependabot.yml b/.github/dependabot.yml
new file mode 100644
index 0000000..17b65ec
--- /dev/null
+++ b/.github/dependabot.yml
@@ -0,0 +1,11 @@
+version: 2
+
+updates:
+ - package-ecosystem: github-actions
+ directory: /
+ schedule:
+ interval: weekly
+ groups:
+ actions:
+ patterns:
+ - "*"
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 70f504e..2a1ba5d 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -7,14 +7,31 @@ on:
branches: [master]
jobs:
+ python-format:
+ runs-on: ubuntu-latest
+ steps:
+ - uses: actions/checkout@v7
+ - name: Check backend formatting
+ uses: astral-sh/ruff-action@v4.1.0
+ with:
+ version: "0.16.0"
+ args: format --check --diff
+ src: backend
+ - name: Check icegraph-client formatting
+ uses: astral-sh/ruff-action@v4.1.0
+ with:
+ version: "0.16.0"
+ args: format --check --diff
+ src: icegraph-client
+
frontend:
runs-on: ubuntu-latest
defaults:
run:
working-directory: frontend
steps:
- - uses: actions/checkout@v4
- - uses: actions/setup-node@v4
+ - uses: actions/checkout@v7
+ - uses: actions/setup-node@v7
with:
node-version: 22
cache: npm
diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml
index 63be792..aa43c97 100644
--- a/.github/workflows/deploy.yml
+++ b/.github/workflows/deploy.yml
@@ -1,9 +1,7 @@
name: Deploy to GitHub Pages
on:
- push:
- tags:
- - "v*"
+ workflow_call:
permissions:
contents: write
@@ -15,10 +13,10 @@ jobs:
steps:
- name: Checkout
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Setup Node.js
- uses: actions/setup-node@v4
+ uses: actions/setup-node@v7
with:
node-version: "24"
cache: "npm"
diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml
index ccac01e..29b8ae3 100644
--- a/.github/workflows/docker-publish.yml
+++ b/.github/workflows/docker-publish.yml
@@ -1,9 +1,12 @@
name: Publish Docker image
on:
- push:
- tags:
- - 'v*'
+ workflow_call:
+ secrets:
+ DOCKERHUB_USERNAME:
+ required: true
+ DOCKERHUB_TOKEN:
+ required: true
jobs:
push_to_registry:
@@ -13,20 +16,20 @@ jobs:
contents: read
steps:
- name: Check out the repo
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
- name: Set up Docker Buildx
- uses: docker/setup-buildx-action@v3
+ uses: docker/setup-buildx-action@v4
- name: Log in to Docker Hub
- uses: docker/login-action@v3
+ uses: docker/login-action@v4
with:
username: ${{ secrets.DOCKERHUB_USERNAME }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
- name: Extract metadata (tags, labels) for Docker
id: meta
- uses: docker/metadata-action@v5
+ uses: docker/metadata-action@v6
with:
images: ${{ secrets.DOCKERHUB_USERNAME }}/icegraph
tags: |
@@ -34,7 +37,7 @@ jobs:
type=raw,value=latest
- name: Build and push Docker image
- uses: docker/build-push-action@v6
+ uses: docker/build-push-action@v7
with:
context: .
push: true
diff --git a/.github/workflows/publish-icegraph-client.yml b/.github/workflows/publish-icegraph-client.yml
index 7e09f43..1e4da82 100644
--- a/.github/workflows/publish-icegraph-client.yml
+++ b/.github/workflows/publish-icegraph-client.yml
@@ -1,22 +1,25 @@
name: Publish icegraph-client to PyPI
on:
- push:
- tags:
- - "v*"
+ workflow_call:
+ secrets:
+ PYPI_API_TOKEN:
+ required: true
jobs:
publish:
name: Build and publish to PyPI
runs-on: ubuntu-latest
+ permissions:
+ contents: read
steps:
- name: Check out the repo
- uses: actions/checkout@v4
+ uses: actions/checkout@v7
with:
fetch-depth: 0
- name: Install uv
- uses: astral-sh/setup-uv@v3
+ uses: astral-sh/setup-uv@v7
- name: Build
working-directory: icegraph-client
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
new file mode 100644
index 0000000..4145cc5
--- /dev/null
+++ b/.github/workflows/release.yml
@@ -0,0 +1,79 @@
+name: Release
+
+on:
+ push:
+ tags:
+ - "v*"
+
+permissions: {}
+
+jobs:
+ docker:
+ name: Docker image
+ uses: ./.github/workflows/docker-publish.yml
+ permissions:
+ contents: read
+ secrets:
+ DOCKERHUB_USERNAME: ${{ secrets.DOCKERHUB_USERNAME }}
+ DOCKERHUB_TOKEN: ${{ secrets.DOCKERHUB_TOKEN }}
+
+ pypi:
+ name: Python client
+ uses: ./.github/workflows/publish-icegraph-client.yml
+ permissions:
+ contents: read
+ secrets:
+ PYPI_API_TOKEN: ${{ secrets.PYPI_API_TOKEN }}
+
+ pages:
+ name: Frontend demo
+ uses: ./.github/workflows/deploy.yml
+ permissions:
+ contents: write
+
+ release-notes:
+ name: Write release notes
+ runs-on: ubuntu-latest
+ needs: [docker, pypi, pages]
+ permissions:
+ contents: write
+ steps:
+ - name: Build notes and publish release
+ env:
+ GH_TOKEN: ${{ github.token }}
+ REPO: ${{ github.repository }}
+ TAG: ${{ github.ref_name }}
+ DOCKER_IMAGE: yanivzalach/icegraph
+ DEMO_URL: https://yanivzalach.github.io/IceGraph/
+ run: |
+ set -euo pipefail
+
+ VERSION="${TAG#v}"
+ NOTES_FILE="$RUNNER_TEMP/release-notes.md"
+
+ GENERATED="$(gh api "repos/$REPO/releases/generate-notes" -f tag_name="$TAG" --jq .body)"
+ CONTRIBUTORS="$(printf '%s\n' "$GENERATED" | grep -oE '@[A-Za-z0-9-]+' | sort -u | paste -sd , - | sed 's/,/, /g' || true)"
+
+ cat > "$NOTES_FILE" <
[PyPI](https://pypi.org/project/icegraph-client/$VERSION/) |
+ | 🌐 Live demo | [$DEMO_URL]($DEMO_URL) |
+ | 📖 Docs | [${DEMO_URL}docs](${DEMO_URL}docs) |
+
+ EOF
+
+ printf '%s\n' "$GENERATED" >> "$NOTES_FILE"
+
+ if [ -n "$CONTRIBUTORS" ]; then
+ printf '\n## Contributors\n\nThanks to %s 🧊\n' "$CONTRIBUTORS" >> "$NOTES_FILE"
+ fi
+
+ if gh release view "$TAG" --repo "$REPO" >/dev/null 2>&1; then
+ gh release edit "$TAG" --repo "$REPO" --notes-file "$NOTES_FILE" --latest
+ else
+ gh release create "$TAG" --repo "$REPO" --title "IceGraph $TAG" --notes-file "$NOTES_FILE" --latest
+ fi
diff --git a/ARCHITECTURE_PHILOSOPHY.md b/ARCHITECTURE_PHILOSOPHY.md
index 8920a28..b265ff0 100644
--- a/ARCHITECTURE_PHILOSOPHY.md
+++ b/ARCHITECTURE_PHILOSOPHY.md
@@ -12,7 +12,7 @@ Introduce no service or infrastructure component unless the feature is impossibl
## 3. Simple deployment
-`SPARK_REMOTE` is the only mandatory configuration. All other tunables default in `backend/constants.py`, overridable via environment variable. One Docker image serves both frontend and backend; no second service to provision or wire up.
+`SPARK_REMOTE` is the only mandatory configuration. All other tunables are defined in `backend/env.py` with defaults and can be overridden via environment variables. One Docker image serves both frontend and backend; no second service to provision or wire up.
## 4. The process is disposable
diff --git a/CLAUDE.md b/CLAUDE.md
index 4d188e2..8301659 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -24,7 +24,7 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
- Python files must not exceed 400 lines.
- Never run git commands — only the user does.
- When `icegraph-client`'s public API or CLI commands change, update the CLI section in `frontend/src/pages/DocsPage.jsx` and the Python Client & CLI bullet in `README.md` to match.
-- Whenever creating a constant that is settable via an environment variable, define its default value in `backend/constants.py` and add its description to `README.md`.
+- Whenever creating a setting that is configurable via an environment variable, define it in `backend/env.py`.
- Frontend work has its own rules, philosophy, and styling conventions in [frontend/CLAUDE.md](frontend/CLAUDE.md) and [frontend/PHILOSOPHY.md](frontend/PHILOSOPHY.md).
## Project Overview
@@ -116,7 +116,7 @@ Backend environment variables (set in `backend/.env`):
| `MAX_SNAPSHOTS_TO_COMPUTE` | 50 | Max snapshots processed per job |
| `MAX_DATA_FILES_TO_COLLECT` | 5000 | Data file iteration limit |
| `MAX_NUMBER_OF_GRAPHS_TO_COMPUTE` | 15 | Concurrent job limit |
-| `MAX_SNAPSHOTS_TO_SHOW` | 2000 | Snapshot selection UI limit |
+| `MAX_SNAPSHOTS_TO_SHOW` | 20 | Snapshot selection UI limit |
| `INCLUDE_NONE_ICEBERG_CATALOGS` | `true` | Include non-Iceberg catalogs (e.g. Spark session catalog) in `/api/v1/tables` |
| `TABLE_LIST_CACHE_TTL_SECONDS` | 60 | Cache TTL for table list endpoint |
diff --git a/README.md b/README.md
index 1462f78..3de742b 100644
--- a/README.md
+++ b/README.md
@@ -93,6 +93,11 @@ cd backend
uv sync
```
+```bash
+cd icegraph-client
+uv sync
+```
+
```bash
cd frontend
npm i
@@ -106,18 +111,7 @@ We will create an `.env` file in the root of the backend directory:
SPARK_REMOTE=sc://localhost:15002 # Our local testing spark, If you use docker, change it to your ip.
```
-If you want to change the default values of the application, you can set the following environment variables:
-
-- `MAX_NUMBER_OF_GRAPHS_TO_COMPUTE`: The maximum number of graphs to compute in parallel. Default is 15.
-- `MAX_SNAPSHOTS_TO_SHOW`: The maximum number of snapshots to show in the snapshot selection page. Default is 2000.
-- `COMPUTE_CLEANUP_TIME_SECONDS`: The time to wait before cleaning up the computed graphs. Default is 12.
-- `MAX_DATA_FILES_TO_COLLECT`: The maximum number of data files to collect. Default is 5000.
-- `MAX_SNAPSHOTS_TO_COMPUTE`: The maximum number of snapshots to compute. Default is 50.
-- `MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS`: The time to wait before forcing an exit of the application when shutting it down. Default is 10.
-- `INCLUDE_NONE_ICEBERG_CATALOGS`: Whether to include catalogs that are not exclusively Iceberg (such as the Spark session catalog) in the table selection autocompletion. Default is true.
-- `TABLE_LIST_CACHE_TTL_SECONDS`: The cache time-to-live for table selection autocompletion endpoint before refresh. Default is 60.
-- `PRODUCTION_MODE`: Whether to serve the app with the Waitress production WSGI server instead of Flask's development server. Always single-process (multi-threaded) so the in-memory job-polling state stays consistent. Default is false; the Docker image sets this to true.
-- `WSGI_THREADS`: The number of threads Waitress uses to handle concurrent requests when `PRODUCTION_MODE` is enabled. Default is 20.
+The supported environment variables, their defaults, and their descriptions are defined in the [`Env` class](backend/env.py).
### 3. Run
@@ -136,9 +130,29 @@ Go to `http://localhost:3000` and explore your tables.
### 4. Before Every Commit
-CI fails the build on any violation, so run all three from the `frontend` directory before you commit:
+CI checks Python formatting and the frontend toolchain. Format each Python project from its own directory.
+
+Backend:
```bash
+cd backend
+
+uv run ruff format .
+```
+
+Python client:
+
+```bash
+cd icegraph-client
+
+uv run ruff format .
+```
+
+Run all three frontend checks from the `frontend` directory:
+
+```bash
+cd frontend
+
npm run format
npm run lint
npm run typecheck
diff --git a/backend/base_classes/base_file.py b/backend/base_classes/base_file.py
index 9cf2e22..2f4b303 100644
--- a/backend/base_classes/base_file.py
+++ b/backend/base_classes/base_file.py
@@ -1,5 +1,5 @@
-from dataclasses import dataclass, fields
-from typing import List
+from dataclasses import dataclass, field, fields
+from typing import List, Optional
from constants import FileType
@@ -14,6 +14,8 @@ class BaseFile:
type: FileType
file_path: str
child_files: List[str]
+ error: Optional[str] = field(default=None, kw_only=True)
+ warning: Optional[str] = field(default=None, kw_only=True)
def to_dict(self):
result_dict = {field.name: getattr(self, field.name) for field in fields(self) if not isinstance(getattr(self, field.name), HiddenFile)}
diff --git a/backend/collectors/collect_data_files.py b/backend/collectors/collect_data_files.py
index 9b41b92..2dead6e 100644
--- a/backend/collectors/collect_data_files.py
+++ b/backend/collectors/collect_data_files.py
@@ -38,19 +38,16 @@ def __init__(
):
super().__init__(full_table_name)
self._manifests = manifests
- self._errors: Dict[str, str] = {}
self._data_files: List[DataFileRecord] = []
@timed
def collect(self) -> FilesCollection:
- data_files_extraction_result = DataFilesExtractor(self._table_name, self._manifests).extract_dataframe()
- self._errors = data_files_extraction_result.errors
+ data_files_rows = DataFilesExtractor(self._table_name, self._manifests).extract_dataframe().collect()
- data_files_rows = data_files_extraction_result.dataframe.collect()
self._data_files = [self._process_data_file_row(data_file_row) for data_file_row in data_files_rows]
- return FilesCollection(files=self._data_files, errors=self._errors)
+ return FilesCollection(files=self._data_files)
def _process_data_file_row(self, data_file_row) -> DataFileRecord:
data_file_dict = data_file_row.asDict(recursive=True)
@@ -59,7 +56,7 @@ def _process_data_file_row(self, data_file_row) -> DataFileRecord:
type=self._detect_file_type(data_file_dict["content"]),
file_path=data_file_dict["file_path"],
format=data_file_dict["file_format"],
- size_gb=f"{(data_file_dict['file_size_in_bytes'] / 1024 ** 3):.10f}",
+ size_gb=f"{(data_file_dict['file_size_in_bytes'] / 1024**3):.10f}",
row_count=data_file_dict["record_count"],
partition=format_partition(data_file_dict["partition"]),
earliest_appearing_snapshot_id=data_file_dict["earliest_snapshot_id"],
diff --git a/backend/collectors/collect_manifests.py b/backend/collectors/collect_manifests.py
index 35d907e..b104e1b 100644
--- a/backend/collectors/collect_manifests.py
+++ b/backend/collectors/collect_manifests.py
@@ -1,5 +1,5 @@
from dataclasses import dataclass
-from typing import Dict, List, Optional
+from typing import List, Optional
import pyspark
@@ -37,19 +37,16 @@ def __init__(
super().__init__(full_table_name)
self._snapshots = snapshots
self._manifests_to_ignore_df = manifests_to_ignore_df
- self._errors: Dict[str, str] = {}
self._manifests: List[ManifestRecord] = []
@timed
def collect(self) -> FilesCollection:
- manifest_extraction_result = ManifestsExtractor(self._table_name, self._snapshots, self._manifests_to_ignore_df).extract_dataframe()
- self._errors = manifest_extraction_result.errors
+ manifests_rows = ManifestsExtractor(self._table_name, self._snapshots, self._manifests_to_ignore_df).extract_dataframe().collect()
- manifests_rows = manifest_extraction_result.dataframe.collect()
self._manifests = [self._process_manifest_row(manifest_row) for manifest_row in manifests_rows]
- return FilesCollection(files=self._manifests, errors=self._errors)
+ return FilesCollection(files=self._manifests)
@staticmethod
def _process_manifest_row(manifest_row) -> ManifestRecord:
diff --git a/backend/collectors/collect_metadata.py b/backend/collectors/collect_metadata.py
index 93369f5..5b397d1 100644
--- a/backend/collectors/collect_metadata.py
+++ b/backend/collectors/collect_metadata.py
@@ -1,6 +1,7 @@
from base_classes.utils import column_to_string_utc
import json
from dataclasses import dataclass
+from functools import cached_property
from typing import Any, Dict, List, Optional
import pyspark.sql
@@ -18,17 +19,17 @@
@dataclass
class MetadataFileRecord(BaseFile):
- timestamp: str
+ timestamp: Optional[str]
snapshot_id: Optional[int]
previous_file: Optional[str]
last_sequence_number: Optional[int]
- partition_spec_id: int
- current_schema_id: int
- sort_order_id: int
+ partition_spec_id: Optional[int]
+ current_schema_id: Optional[int]
+ sort_order_id: Optional[int]
refs: Dict[str, Any]
properties: Dict[str, str]
pointed_snapshots_files: Optional[List[Dict[str, str]]]
- pointed_metadata_log_count: int
+ pointed_metadata_log_count: Optional[int]
class CollectMetadata(Collector):
@@ -45,22 +46,26 @@ def __init__(
self._start_metadata_cutoff = start_metadata_cutoff
self._end_metadata_cutoff = end_metadata_cutoff
+ self._ordered_metadata_to_timestamp: Dict[str, str] = {}
+
self._metadata_files: List[MetadataFileRecord] = []
+ self._bad_metadata_files: List[MetadataFileRecord] = []
self._errors: Dict[str, str] = {}
@timed
def collect(self) -> FilesCollection:
try:
- metadata_files = self._query_metadata_files()
- metadata_files_df = self._build_metadata_files_df(metadata_files)
+ self._ordered_metadata_to_timestamp = self._query_metadata_files()
+ metadata_files_df = self._build_metadata_files_df()
if metadata_files_df is not None:
snap_id_to_path = self._get_snap_id_to_path()
- rows = metadata_files_df.orderBy(F.desc("metadata_timestamp")).collect()
- self._metadata_files = [
- self._parse_metadata_row(index, row.asDict(recursive=True), rows, snap_id_to_path) for index, row in enumerate(rows)
- ]
+ for row in metadata_files_df.collect():
+ self._metadata_files.append(self._parse_metadata_row(row.asDict(recursive=True), snap_id_to_path))
+
+ self._metadata_files.extend(self._bad_metadata_files)
+ self._apply_metadata_order()
except Exception as e:
logger.error(f"[{self._table_name}] metadata collection failed", exc_info=True)
@@ -68,6 +73,10 @@ def collect(self) -> FilesCollection:
return FilesCollection(files=self._metadata_files, errors=self._errors)
+ @cached_property
+ def _ordered_metadata_paths(self) -> list[str]:
+ return list(self._ordered_metadata_to_timestamp.keys())
+
def _query_metadata_files(self) -> dict:
metadata_df = (
self._spark.sql(f"SELECT * FROM {self._table_name}.metadata_log_entries")
@@ -75,13 +84,14 @@ def _query_metadata_files(self) -> dict:
.select("file", "metadata_timestamp")
.filter(F.col("metadata_timestamp") >= F.lit(str(self._start_metadata_cutoff)))
.filter(F.col("metadata_timestamp") <= F.lit(str(self._end_metadata_cutoff)))
+ .orderBy(F.desc("metadata_timestamp"))
.withColumn("metadata_timestamp", column_to_string_utc("metadata_timestamp"))
)
return {row.file: row.metadata_timestamp for row in metadata_df.collect()}
- def _build_metadata_files_df(self, metadata_files: dict) -> Optional[pyspark.sql.DataFrame]:
+ def _build_metadata_files_df(self) -> Optional[pyspark.sql.DataFrame]:
metadata_files_df = None
- for file, timestamp in metadata_files.items():
+ for file, timestamp in self._ordered_metadata_to_timestamp.items():
try:
df = get_metadata_row_slim_df_from_path(file).withColumn("metadata_timestamp", F.lit(timestamp)).withColumn("file", F.lit(file))
metadata_files_df = df if metadata_files_df is None else metadata_files_df.unionByName(df, allowMissingColumns=True)
@@ -91,13 +101,45 @@ def _build_metadata_files_df(self, metadata_files: dict) -> Optional[pyspark.sql
f"[{self._table_name}] Metadata file read error for {file}",
exc_info=True,
)
- self._errors[file] = f"Metadata file read error: {e}"
+ self._bad_metadata_files.append(
+ MetadataFileRecord(
+ type=FileType.METADATA,
+ file_path=file,
+ child_files=[],
+ error=str(e),
+ timestamp=timestamp,
+ snapshot_id=None,
+ previous_file=self._get_previous_metadata_file(file),
+ last_sequence_number=None,
+ partition_spec_id=None,
+ current_schema_id=None,
+ sort_order_id=None,
+ refs={},
+ properties={},
+ pointed_snapshots_files=None,
+ pointed_metadata_log_count=None,
+ )
+ )
return metadata_files_df
+ def _apply_metadata_order(self) -> None:
+ if not self._metadata_files:
+ return
+
+ metadata_file_by_path = {metadata_file.file_path: metadata_file for metadata_file in self._metadata_files}
+
+ self._metadata_files = [metadata_file_by_path[file_path] for file_path in self._ordered_metadata_paths]
+ self._metadata_files[0].type = FileType.MAIN_METADATA
+
def _get_snap_id_to_path(self) -> Dict[int, str]:
return {s.snapshot_id: s.file_path for s in (self._snapshots or [])}
+ def _get_previous_metadata_file(self, file_path: str) -> Optional[str]:
+ older_index = self._ordered_metadata_paths.index(file_path) + 1
+
+ return self._ordered_metadata_paths[older_index] if older_index < len(self._ordered_metadata_paths) else None
+
@staticmethod
def _parse_refs(row: dict) -> dict:
return json.loads(row["refs"]) if row.get("refs") else {}
@@ -115,10 +157,7 @@ def _build_branches_child_files(refs: dict, snap_id_to_path: dict) -> List[str]:
return branches_child_files
- def _parse_metadata_row(self, index: int, row: dict, rows: list, snap_id_to_path: dict) -> MetadataFileRecord:
- number_of_rows = len(rows)
- file_type = FileType.MAIN_METADATA if index == 0 else FileType.METADATA
-
+ def _parse_metadata_row(self, row: dict, snap_id_to_path: dict) -> MetadataFileRecord:
refs = self._parse_refs(row)
branches_child_files = self._build_branches_child_files(refs, snap_id_to_path)
@@ -126,11 +165,11 @@ def _parse_metadata_row(self, index: int, row: dict, rows: list, snap_id_to_path
child_files = ([current_snap_path] if current_snap_path else []) + branches_child_files
return MetadataFileRecord(
- type=file_type,
+ type=FileType.METADATA,
file_path=row["file"],
timestamp=str(row["metadata_timestamp"]),
snapshot_id=row["current-snapshot-id"],
- previous_file=(rows[index + 1]["file"] if index + 1 < number_of_rows else None),
+ previous_file=self._get_previous_metadata_file(row["file"]),
last_sequence_number=(row["last-sequence-number"] if "last-sequence-number" in row else None),
partition_spec_id=row["default-spec-id"],
current_schema_id=row["current-schema-id"],
diff --git a/backend/collectors/collect_snapshots.py b/backend/collectors/collect_snapshots.py
index 461a5f2..21eb402 100644
--- a/backend/collectors/collect_snapshots.py
+++ b/backend/collectors/collect_snapshots.py
@@ -1,5 +1,3 @@
-from base_classes.utils import column_to_string_utc
-import os
from dataclasses import dataclass
from typing import Dict, List, Optional
@@ -8,12 +6,11 @@
from pyspark.sql import functions as F
from base_classes.base_file import BaseFile
+from base_classes.utils import column_to_string_utc, timed
from collectors.collector import Collector, FilesCollection
-from constants import FileType, MAX_SNAPSHOTS_TO_COMPUTE
+from constants import FileType
+from env import Env
from icegraph_logger import logger
-from base_classes.utils import timed
-
-max_snapshots_to_compute = int(os.getenv("MAX_SNAPSHOTS_TO_COMPUTE", MAX_SNAPSHOTS_TO_COMPUTE))
@dataclass
@@ -63,12 +60,12 @@ def _query_snapshots_df(self) -> pyspark.sql.DataFrame:
@staticmethod
def _validate_snapshot_count(snapshots_df: pyspark.sql.DataFrame) -> None:
- if snapshots_df.count() > max_snapshots_to_compute:
- raise ValueError(f"Too many snapshots to compute. Maximum is {max_snapshots_to_compute}.")
+ if snapshots_df.count() > Env.MAX_SNAPSHOTS_TO_COMPUTE:
+ raise ValueError(f"Too many snapshots to compute. Maximum is {Env.MAX_SNAPSHOTS_TO_COMPUTE}.")
@staticmethod
def _format_summary(summary: dict) -> dict:
- return {k: (f"{(int(v) / (1024 ** 3)):.5f} GB" if k.endswith("files-size") else v) for k, v in summary.items()}
+ return {k: (f"{(int(v) / (1024**3)):.5f} GB" if k.endswith("files-size") else v) for k, v in summary.items()}
def _parse_snapshot_row(self, snapshot) -> SnapshotRecord:
return SnapshotRecord(
diff --git a/backend/collectors/collector.py b/backend/collectors/collector.py
index 1450467..70f5abe 100644
--- a/backend/collectors/collector.py
+++ b/backend/collectors/collector.py
@@ -10,7 +10,6 @@
class FilesCollection:
files: List[BaseFile] = field(default_factory=list)
errors: Dict[str, str] = field(default_factory=dict)
- warnings: Dict[str, str] = field(default_factory=dict)
class Collector(SparkTableAction, ABC):
diff --git a/backend/constants.py b/backend/constants.py
index d308b24..ddd5bc6 100644
--- a/backend/constants.py
+++ b/backend/constants.py
@@ -1,28 +1,8 @@
import inspect
from enum import Enum
-MAX_NUMBER_OF_GRAPHS_TO_COMPUTE = 15
-
-MAX_SNAPSHOTS_TO_SHOW = 20
-
-MAX_SNAPSHOTS_TO_COMPUTE = 50
-
-COMPUTE_CLEANUP_TIME_SECONDS = 12
-
-MAX_DATA_FILES_TO_COLLECT = 5_000
-
-TABLE_LIST_CACHE_TTL_SECONDS = 60
-
-INCLUDE_NONE_ICEBERG_CATALOGS = "true"
-
-MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS = 10
-
APPLICATION_PORT = 5_050
-PRODUCTION_MODE = "false"
-
-WSGI_THREADS = 20
-
MAIN_BRANCH_ICEBERG_TABLE_NAME = "main"
JOB_TOKEN_FIELD = "X-IceGraph-Job-Token"
@@ -53,3 +33,7 @@ class FileType(Enum):
unless a newer visible snapshot also references them, in which case they are included.
Every data file you see is referenced by at least one snapshot that is newer than the cut-off snapshot.
""")
+
+DATA_FILES_CUTOFF_MANIFEST_WARNING = inspect.cleandoc("""
+The data files of the manifest were not loaded/attached because the limit of {max_data_files_to_collect} data files was reached.
+""")
diff --git a/backend/env.py b/backend/env.py
new file mode 100644
index 0000000..13d6590
--- /dev/null
+++ b/backend/env.py
@@ -0,0 +1,39 @@
+import os
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+class Env:
+ """Application settings loaded from environment variables."""
+
+ # Maximum number of graphs to compute in parallel.
+ MAX_NUMBER_OF_GRAPHS_TO_COMPUTE = int(os.getenv("MAX_NUMBER_OF_GRAPHS_TO_COMPUTE", "15"))
+
+ # Maximum number of snapshots shown on the snapshot selection page.
+ MAX_SNAPSHOTS_TO_SHOW = int(os.getenv("MAX_SNAPSHOTS_TO_SHOW", "20"))
+
+ # Maximum number of snapshots processed per graph.
+ MAX_SNAPSHOTS_TO_COMPUTE = int(os.getenv("MAX_SNAPSHOTS_TO_COMPUTE", "50"))
+
+ # Delay before a completed graph is removed from memory.
+ COMPUTE_CLEANUP_TIME_SECONDS = int(os.getenv("COMPUTE_CLEANUP_TIME_SECONDS", "12"))
+
+ # Maximum number of data files collected per graph.
+ MAX_DATA_FILES_TO_COLLECT = int(os.getenv("MAX_DATA_FILES_TO_COLLECT", "5000"))
+
+ # Cache lifetime for the table selection endpoint.
+ TABLE_LIST_CACHE_TTL_SECONDS = int(os.getenv("TABLE_LIST_CACHE_TTL_SECONDS", "60"))
+
+ # Whether non-Iceberg catalogs are included in table selection.
+ INCLUDE_NONE_ICEBERG_CATALOGS = os.getenv("INCLUDE_NONE_ICEBERG_CATALOGS", "true").lower() == "true"
+
+ # Maximum time allowed for graceful application shutdown.
+ MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS = int(os.getenv("MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS", "10"))
+
+ # Whether to serve the application with Waitress.
+ PRODUCTION_MODE = os.getenv("PRODUCTION_MODE", "false").lower() == "true"
+
+ # Number of request threads used by Waitress.
+ WSGI_THREADS = int(os.getenv("WSGI_THREADS", "20"))
diff --git a/backend/extractors/data_files_extractor.py b/backend/extractors/data_files_extractor.py
index 7eb2ad5..398ecb1 100644
--- a/backend/extractors/data_files_extractor.py
+++ b/backend/extractors/data_files_extractor.py
@@ -1,14 +1,13 @@
-import os
from typing import List
-from pyspark.sql import Window, functions as F
+import pyspark
+from pyspark.sql import Window
+from pyspark.sql import functions as F
from pyspark.sql.types import LongType, StringType, StructField, StructType
from collectors.collect_manifests import ManifestRecord
-from constants import MAX_DATA_FILES_TO_COLLECT
-from extractors.extractor import ExtractionResult, Extractor
-
-max_data_files_to_collect = int(os.getenv("MAX_DATA_FILES_TO_COLLECT", MAX_DATA_FILES_TO_COLLECT))
+from env import Env
+from extractors.extractor import Extractor
DATA_FILE_RECORD_SCHEMA = StructType(
[
@@ -42,9 +41,8 @@ class DataFilesExtractor(Extractor):
def __init__(self, table_name: str, manifest_entries: List[ManifestRecord]):
super().__init__(table_name)
self._manifest_entries = manifest_entries
- self._errors = {}
- def extract_dataframe(self) -> ExtractionResult:
+ def extract_dataframe(self) -> pyspark.sql.DataFrame:
data_files_df = self._collect_data_files_from_manifests(self._manifest_entries)
data_files_with_latest_ts_df = self._match_data_file_to_latest_snapshot(data_files_df)
@@ -55,9 +53,7 @@ def extract_dataframe(self) -> ExtractionResult:
snapshot_timestamp_cutoff_df = self._find_cutoff_snapshot_timestamp(data_files_limited_df)
- included_data_files_df = self._find_included_data_files(data_files_limited_df, snapshot_timestamp_cutoff_df)
-
- return ExtractionResult(included_data_files_df, self._errors)
+ return self._find_included_data_files(data_files_limited_df, snapshot_timestamp_cutoff_df)
@staticmethod
def _group_data_files_by_manifests(avro_df):
@@ -114,7 +110,7 @@ def _join_data_file_with_manifest_entries(latest_df, manifest_entries_df):
@staticmethod
def _limit_and_rank_files_by_snapshot_timestamp(df):
- df = df.orderBy(F.desc("latest_snapshot_timestamp")).limit(max_data_files_to_collect + 1)
+ df = df.orderBy(F.desc("latest_snapshot_timestamp")).limit(Env.MAX_DATA_FILES_TO_COLLECT + 1)
row_num_window = Window.orderBy(F.desc("latest_snapshot_timestamp"))
df = df.withColumn("row_num", F.row_number().over(row_num_window))
@@ -124,7 +120,7 @@ def _limit_and_rank_files_by_snapshot_timestamp(df):
@staticmethod
def _find_cutoff_snapshot_timestamp(df):
return (
- df.filter(F.col("row_num") == max_data_files_to_collect + 1)
+ df.filter(F.col("row_num") == Env.MAX_DATA_FILES_TO_COLLECT + 1)
.agg(F.coalesce(F.first("latest_snapshot_timestamp"), F.lit(0).cast("timestamp")).alias("snapshot_timestamp_cutoff"))
.select("snapshot_timestamp_cutoff")
)
@@ -140,16 +136,14 @@ def _find_included_data_files(grouped_files_limited_df, snapshot_timestamp_cutof
def _collect_data_files_from_manifests(self, manifest_rows):
avro_df = None
for manifest_entry in manifest_rows:
- try:
- df = self._collect_data_files_from_manifest(manifest_entry)
-
- if avro_df is None:
- avro_df = df
- else:
- avro_df = avro_df.unionByName(df, allowMissingColumns=True)
-
- except Exception as e:
- self._errors[manifest_entry.file_path] = f"Avro read error: {e}"
+ df = self._read_source(manifest_entry, lambda: self._collect_data_files_from_manifest(manifest_entry))
+ if df is None:
+ continue
+
+ if avro_df is None:
+ avro_df = df
+ else:
+ avro_df = avro_df.unionByName(df, allowMissingColumns=True)
if avro_df is None:
return self._spark.createDataFrame([], DATA_FILE_RECORD_SCHEMA)
diff --git a/backend/extractors/extractor.py b/backend/extractors/extractor.py
index 162a837..fcfbd75 100644
--- a/backend/extractors/extractor.py
+++ b/backend/extractors/extractor.py
@@ -1,19 +1,27 @@
from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-from typing import Dict
+from typing import Callable, Optional
import pyspark
+from base_classes.base_file import BaseFile
from base_classes.spark_table_action import SparkTableAction
-
-
-@dataclass(frozen=True)
-class ExtractionResult:
- dataframe: pyspark.sql.DataFrame
- errors: Dict[str, str] = field(default_factory=dict)
+from icegraph_logger import logger
class Extractor(SparkTableAction, ABC):
@abstractmethod
- def extract_dataframe(self) -> ExtractionResult:
+ def extract_dataframe(self) -> pyspark.sql.DataFrame:
pass
+
+ def _read_source(self, source_file: BaseFile, read_source: Callable[[], pyspark.sql.DataFrame]) -> Optional[pyspark.sql.DataFrame]:
+ try:
+ data = read_source()
+ data.schema # Trigger the file metadata read
+
+ return data
+
+ except Exception as e:
+ logger.error(f"[{self._table_name}] Failed to read file {source_file.file_path}", exc_info=True)
+ source_file.error = str(e)
+
+ return None
diff --git a/backend/extractors/manifests_extractor.py b/backend/extractors/manifests_extractor.py
index 2af85b2..ba6d25c 100644
--- a/backend/extractors/manifests_extractor.py
+++ b/backend/extractors/manifests_extractor.py
@@ -3,7 +3,7 @@
from pyspark.sql.types import LongType, StringType, StructField, StructType
from collectors.collect_snapshots import SnapshotRecord
-from extractors.extractor import ExtractionResult, Extractor
+from extractors.extractor import Extractor
SNAPSHOT_TO_TIMESTAMP_SCHEMA = StructType(
[
@@ -31,34 +31,29 @@ def __init__(
super().__init__(table_name)
self._snapshots = snapshots
self._manifests_to_ignore_df = manifests_to_ignore_df
- self._errors = {}
- def extract_dataframe(self) -> ExtractionResult:
+ def extract_dataframe(self) -> pyspark.sql.DataFrame:
manifests_df = self._union_manifests_for_snapshots()
manifests_with_timestamps_df = self._enrich_manifests_with_timestamps(manifests_df)
valid_manifests_df = self._filter_ignored_manifests(manifests_with_timestamps_df)
- manifests_df = self._aggregate_snapshots_by_manifests_sorted(valid_manifests_df)
- return ExtractionResult(
- manifests_df,
- self._errors,
- )
+ return self._aggregate_snapshots_by_manifests_sorted(valid_manifests_df)
def _union_manifests_for_snapshots(self) -> pyspark.sql.DataFrame:
result = None
for snapshot in self._snapshots:
snap_id = snapshot.snapshot_id
manifest_list_path = snapshot.file_path
- try:
- df = self._read_manifests_for_snapshot(manifest_list_path, snap_id)
- if result is None:
- result = df
- else:
- result = result.unionByName(df, allowMissingColumns=True)
-
- except Exception as e:
- self._errors[manifest_list_path] = f"Failed to read/union manifest list: {e}"
+
+ df = self._read_source(snapshot, lambda: self._read_manifests_for_snapshot(manifest_list_path, snap_id))
+ if df is None:
+ continue
+
+ if result is None:
+ result = df
+ else:
+ result = result.unionByName(df, allowMissingColumns=True)
if result is None:
result = self._spark.createDataFrame([], MANIFEST_BASE_SCHEMA)
diff --git a/backend/graph_normalizer/graph_normalizer.py b/backend/graph_normalizer/graph_normalizer.py
index 79cab66..68fc815 100644
--- a/backend/graph_normalizer/graph_normalizer.py
+++ b/backend/graph_normalizer/graph_normalizer.py
@@ -3,11 +3,10 @@
class GraphNormalizer:
-
def __init__(self, table_data: TableInventoryResult):
self._files = table_data.metadata_files + table_data.snapshots + table_data.manifests + table_data.data_files
- self._errors = table_data.errors
- self._warnings = table_data.warnings
+ self._table_errors = table_data.errors
+ self._table_warnings = table_data.warnings
self._current_table_metadata = table_data.current_table_specs
def normalize(self):
@@ -17,7 +16,7 @@ def normalize(self):
{
"nodes": nodes,
"metadata": self._current_table_metadata,
- "errors": self._errors,
- "warnings": self._warnings,
+ "errors": self._table_errors,
+ "warnings": self._table_warnings,
}
)
diff --git a/backend/main.py b/backend/main.py
index b233cbc..de78e1c 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -6,46 +6,27 @@
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
-from dotenv import load_dotenv
from flask import Flask, jsonify, request, send_from_directory
from pyspark.errors import AnalysisException
-from constants import (
- APPLICATION_PORT,
- COMPUTE_CLEANUP_TIME_SECONDS,
- JOB_TOKEN_FIELD,
- MAX_NUMBER_OF_GRAPHS_TO_COMPUTE,
- MAX_SNAPSHOTS_TO_SHOW,
- MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS,
- INCLUDE_NONE_ICEBERG_CATALOGS,
- PRODUCTION_MODE,
- WSGI_THREADS,
-)
-from spark_connect import close_spark_connect_session
+from base_classes.utils import verify_iceberg_table
+from constants import APPLICATION_PORT, JOB_TOKEN_FIELD
+from env import Env
from graph_normalizer.graph_normalizer import GraphNormalizer
from icegraph_logger import logger
from snapshot_analyzer.snapshot_analyzer import SnapshotAnalyzer
from snapshot_map.snapshot_mapping import collect_snapshot_map
-from table_list_catalog.table_list_catalog import TableListCatalog
+from spark_connect import close_spark_connect_session
from table_inventory.table_inventory import TableInventory
-from base_classes.utils import verify_iceberg_table
+from table_list_catalog.table_list_catalog import TableListCatalog
-load_dotenv()
app = Flask(__name__, static_url_path="/static")
app.json.sort_keys = False
job_lock = threading.Lock()
jobs: dict[str, dict] = {}
-max_number_of_graphs_to_compute = int(os.getenv("MAX_NUMBER_OF_GRAPHS_TO_COMPUTE", MAX_NUMBER_OF_GRAPHS_TO_COMPUTE))
-compute_cleanup_time_seconds = int(os.getenv("COMPUTE_CLEANUP_TIME_SECONDS", COMPUTE_CLEANUP_TIME_SECONDS))
-max_snapshots_to_show = int(os.getenv("MAX_SNAPSHOTS_TO_SHOW", MAX_SNAPSHOTS_TO_SHOW))
-max_graceful_shutdown_time_seconds = int(os.getenv("MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS", MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS))
-include_none_iceberg_catalogs = str(os.getenv("INCLUDE_NONE_ICEBERG_CATALOGS", INCLUDE_NONE_ICEBERG_CATALOGS)).lower() == "true"
-production_mode = str(os.getenv("PRODUCTION_MODE", PRODUCTION_MODE)).lower() == "true"
-wsgi_threads = int(os.getenv("WSGI_THREADS", WSGI_THREADS))
-
-executor_pool = ThreadPoolExecutor(max_workers=max_number_of_graphs_to_compute)
+executor_pool = ThreadPoolExecutor(max_workers=Env.MAX_NUMBER_OF_GRAPHS_TO_COMPUTE)
def _safe_update_job(job_id, **fields):
@@ -62,7 +43,7 @@ def _cleanup_job(job_id):
def _schedule_cleanup(job_id, is_in_lock_block=False):
timer = threading.Timer(
- compute_cleanup_time_seconds,
+ Env.COMPUTE_CLEANUP_TIME_SECONDS,
lambda job_id=job_id: _cleanup_job(job_id),
)
timer.daemon = True
@@ -118,7 +99,7 @@ def list_tables():
return jsonify(
{
"tables": tables,
- "include_none_iceberg_catalogs": include_none_iceberg_catalogs,
+ "include_none_iceberg_catalogs": Env.INCLUDE_NONE_ICEBERG_CATALOGS,
}
)
@@ -136,7 +117,7 @@ def snapshot_map(table_name):
try:
verify_iceberg_table(table_name)
- result = collect_snapshot_map(table_name, max_snapshots_to_show)
+ result = collect_snapshot_map(table_name, Env.MAX_SNAPSHOTS_TO_SHOW)
return jsonify(result)
@@ -228,21 +209,21 @@ def get_job_status(job_id):
def _force_exit():
- logger.error(f"Graceful shutdown timed out after {max_graceful_shutdown_time_seconds} seconds - forcing exit.")
+ logger.error(f"Graceful shutdown timed out after {Env.MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS} seconds - forcing exit.")
os._exit(1)
if __name__ == "__main__":
try:
- if production_mode:
+ if Env.PRODUCTION_MODE:
from waitress import serve
- serve(app, host="0.0.0.0", port=APPLICATION_PORT, threads=wsgi_threads)
+ serve(app, host="0.0.0.0", port=APPLICATION_PORT, threads=Env.WSGI_THREADS)
else:
app.run(host="0.0.0.0", port=APPLICATION_PORT, debug=True)
finally:
- watchdog = threading.Timer(max_graceful_shutdown_time_seconds, _force_exit)
+ watchdog = threading.Timer(Env.MAX_GRACEFUL_SHUTDOWN_TIME_SECONDS, _force_exit)
watchdog.daemon = True
watchdog.start()
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index c8d718a..4b7b9db 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -16,5 +16,11 @@ dependencies = [
"setuptools>=75.0.0",
"waitress==3.0.2",
]
-[tool.black]
-line-length = 150
\ No newline at end of file
+
+[dependency-groups]
+dev = [
+ "ruff==0.16.0",
+]
+
+[tool.ruff]
+line-length = 150
diff --git a/backend/search_cutoff/find_search_cutoff.py b/backend/search_cutoff/find_search_cutoff.py
index a247e4b..26ec510 100644
--- a/backend/search_cutoff/find_search_cutoff.py
+++ b/backend/search_cutoff/find_search_cutoff.py
@@ -65,11 +65,15 @@ def _get_start_cutoffs(
table_name: str,
start_snapshot_id: int,
) -> StartCutoffs:
- row = spark.sql(f"""
+ row = (
+ spark.sql(f"""
SELECT committed_at, parent_id
FROM {table_name}.snapshots
WHERE snapshot_id = {start_snapshot_id}
- """).withColumn("committed_at", column_to_string_utc("committed_at")).first()
+ """)
+ .withColumn("committed_at", column_to_string_utc("committed_at"))
+ .first()
+ )
if not row:
return StartCutoffs(
@@ -80,11 +84,15 @@ def _get_start_cutoffs(
snapshot_cutoff = to_arrow_utc(row.committed_at)
- meta_row = spark.sql(f"""
+ meta_row = (
+ spark.sql(f"""
SELECT MIN(timestamp) AS ts
FROM {table_name}.metadata_log_entries
WHERE latest_snapshot_id = {start_snapshot_id}
- """).withColumn("ts", column_to_string_utc("ts")).first()
+ """)
+ .withColumn("ts", column_to_string_utc("ts"))
+ .first()
+ )
metadata_cutoff = to_arrow_utc(meta_row.ts) if meta_row and meta_row.ts else snapshot_cutoff
@@ -102,11 +110,15 @@ def _get_end_cutoffs(
table_name: str,
end_snapshot_id: int,
) -> EndCutoffs:
- row = spark.sql(f"""
+ row = (
+ spark.sql(f"""
SELECT committed_at
FROM {table_name}.snapshots
WHERE snapshot_id = {end_snapshot_id}
- """).withColumn("committed_at", column_to_string_utc("committed_at")).first()
+ """)
+ .withColumn("committed_at", column_to_string_utc("committed_at"))
+ .first()
+ )
if not row:
return EndCutoffs(
@@ -116,11 +128,15 @@ def _get_end_cutoffs(
snapshot_cutoff = to_arrow_utc(row.committed_at)
- meta_row = spark.sql(f"""
+ meta_row = (
+ spark.sql(f"""
SELECT MAX(timestamp) AS ts
FROM {table_name}.metadata_log_entries
WHERE latest_snapshot_id = {end_snapshot_id}
- """).withColumn("ts", column_to_string_utc("ts")).first()
+ """)
+ .withColumn("ts", column_to_string_utc("ts"))
+ .first()
+ )
metadata_cutoff = to_arrow_utc(meta_row.ts) if meta_row and meta_row.ts else snapshot_cutoff
diff --git a/backend/table_inventory/table_inventory.py b/backend/table_inventory/table_inventory.py
index fd6f7eb..8a27621 100644
--- a/backend/table_inventory/table_inventory.py
+++ b/backend/table_inventory/table_inventory.py
@@ -1,4 +1,3 @@
-import os
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
@@ -9,13 +8,12 @@
from collectors.collect_manifests import CollectManifests, ManifestRecord
from collectors.collect_metadata import CollectMetadata, MetadataFileRecord
from collectors.collect_snapshots import CollectSnapshots, SnapshotRecord
-from constants import DATA_FILES_CUTOFF_WARNING, FileType, MAX_DATA_FILES_TO_COLLECT
+from constants import DATA_FILES_CUTOFF_MANIFEST_WARNING, DATA_FILES_CUTOFF_WARNING, FileType
+from env import Env
from icegraph_logger import logger
from search_cutoff.find_search_cutoff import SearchCutoff, find_search_cutoff
from table_inventory.utils import format_schemas_to_full_dict, get_json_metadata_from_path, parse_json_string_fields
-max_data_files_to_collect = int(os.getenv("MAX_DATA_FILES_TO_COLLECT", MAX_DATA_FILES_TO_COLLECT))
-
@dataclass
class TableInventoryResult:
@@ -65,6 +63,7 @@ def build(self):
self._warn_if_data_cutoff_happened()
self._set_current_table_specs()
+ self._collect_file_errors()
return TableInventoryResult(
errors=self._errors,
@@ -92,7 +91,6 @@ def _collect_and_set_snapshots(self):
).collect()
self._errors.update(snapshot_collection.errors)
- self._warnings.update(snapshot_collection.warnings)
self._snapshots = snapshot_collection.files
@@ -105,7 +103,6 @@ def _collect_metadata_manifests_and_data_files(self):
metadata_collection = metadata_future.result()
self._errors.update(metadata_collection.errors)
- self._warnings.update(metadata_collection.warnings)
self._metadata_files = metadata_collection.files
@@ -118,8 +115,6 @@ def _collect_metadata_manifests_and_data_files(self):
self._errors.update(manifests_collection.errors)
self._errors.update(data_files_collection.errors)
- self._warnings.update(manifests_collection.warnings)
- self._warnings.update(data_files_collection.warnings)
self._manifests = manifests_collection.files
self._data_files = data_files_collection.files
@@ -161,7 +156,6 @@ def _attach_snapshot_files_to_manifest_files(self):
for manifest in self._manifests:
for snapshot_id in manifest.hidden_manifest_data.pointing_snapshots:
-
snapshot = snapshot_id_to_snapshot_file_map.get(snapshot_id)
if not snapshot:
self._errors[f"Linking {snapshot_id} -> {manifest.file_path}"] = "Snapshot not found"
@@ -204,18 +198,27 @@ def _warn_if_data_cutoff_happened(self):
max_manifest_added_snapshot_id = None
for manifest in self._manifests:
- if len(manifest.child_files) == 0:
- if max_manifest_added_snapshot_timestamp is None or max_manifest_added_snapshot_timestamp < manifest.added_snapshot_timestamp:
- max_manifest_added_snapshot_timestamp = manifest.added_snapshot_timestamp
- max_manifest_added_snapshot_id = manifest.added_snapshot_id
+ if manifest.child_files or manifest.error or manifest.added_snapshot_timestamp is None:
+ continue
+
+ manifest.warning = DATA_FILES_CUTOFF_MANIFEST_WARNING.format(max_data_files_to_collect=Env.MAX_DATA_FILES_TO_COLLECT)
+
+ if max_manifest_added_snapshot_timestamp is None or max_manifest_added_snapshot_timestamp < manifest.added_snapshot_timestamp:
+ max_manifest_added_snapshot_timestamp = manifest.added_snapshot_timestamp
+ max_manifest_added_snapshot_id = manifest.added_snapshot_id
if max_manifest_added_snapshot_timestamp is not None:
self._warnings["data_files_cutoff"] = DATA_FILES_CUTOFF_WARNING.format(
- max_data_files_to_collect=max_data_files_to_collect,
+ max_data_files_to_collect=Env.MAX_DATA_FILES_TO_COLLECT,
added_snapshot_id=max_manifest_added_snapshot_id,
added_snapshot_timestamp=max_manifest_added_snapshot_timestamp,
)
+ def _collect_file_errors(self):
+ file_groups = (self._metadata_files, self._snapshots, self._manifests, self._data_files)
+ for files in file_groups:
+ self._errors.update({file.file_path: file.error for file in files if file.error})
+
def _set_current_table_specs(self):
self._current_table_specs = {"table-name": self._table_name}
diff --git a/backend/table_list_catalog/table_list_catalog.py b/backend/table_list_catalog/table_list_catalog.py
index 3865a31..57ad4b5 100644
--- a/backend/table_list_catalog/table_list_catalog.py
+++ b/backend/table_list_catalog/table_list_catalog.py
@@ -1,16 +1,19 @@
-from table_list_catalog.utils import get_spark_default_catalog
-import os
import threading
-import arrow
from dataclasses import dataclass
from typing import Optional
+import arrow
+
from base_classes.utils import timed
-from constants import TABLE_LIST_CACHE_TTL_SECONDS
+from env import Env
from spark_connect import open_spark_connect_session
-from table_list_catalog.utils import collect_catalogs_tables_names, collect_databases_in_catalogs, list_catalog_names, filter_catalogs_to_include
-
-table_list_cache_ttl_seconds = int(os.getenv("TABLE_LIST_CACHE_TTL_SECONDS", TABLE_LIST_CACHE_TTL_SECONDS))
+from table_list_catalog.utils import (
+ collect_catalogs_tables_names,
+ collect_databases_in_catalogs,
+ filter_catalogs_to_include,
+ get_spark_default_catalog,
+ list_catalog_names,
+)
@dataclass
@@ -46,7 +49,7 @@ def collect(self) -> list[str]:
@staticmethod
def _fresh_cached_tables() -> Optional[list[str]]:
cache = TableListCatalog._cache
- if cache and (arrow.utcnow() - cache.timestamp).total_seconds() < table_list_cache_ttl_seconds:
+ if cache and (arrow.utcnow() - cache.timestamp).total_seconds() < Env.TABLE_LIST_CACHE_TTL_SECONDS:
return cache.tables
return None
diff --git a/backend/table_list_catalog/utils.py b/backend/table_list_catalog/utils.py
index 7828009..801b6e1 100644
--- a/backend/table_list_catalog/utils.py
+++ b/backend/table_list_catalog/utils.py
@@ -1,13 +1,10 @@
-from typing import List
-from pyspark.sql import functions as F
-import os
from contextlib import suppress
-from typing import Optional
+from typing import List, Optional
from pyspark.sql import SparkSession
-from constants import INCLUDE_NONE_ICEBERG_CATALOGS
+from pyspark.sql import functions as F
-include_none_iceberg_catalogs = str(os.getenv("INCLUDE_NONE_ICEBERG_CATALOGS", INCLUDE_NONE_ICEBERG_CATALOGS)).lower() == "true"
+from env import Env
def get_spark_default_catalog(spark: SparkSession) -> str:
@@ -32,7 +29,7 @@ def filter_catalogs_to_include(spark: SparkSession, catalogs: list[str]) -> list
for catalog in catalogs:
catalog_config_value = get_spark_catalog_config_value(spark, catalog)
- if include_none_iceberg_catalogs or is_iceberg_spark_catalog(catalog_config_value):
+ if Env.INCLUDE_NONE_ICEBERG_CATALOGS or is_iceberg_spark_catalog(catalog_config_value):
catalogs_to_include.append(catalog)
return catalogs_to_include
diff --git a/backend/uv.lock b/backend/uv.lock
index 5bb38f8..6cced5d 100644
--- a/backend/uv.lock
+++ b/backend/uv.lock
@@ -146,6 +146,11 @@ dependencies = [
{ name = "waitress" },
]
+[package.dev-dependencies]
+dev = [
+ { name = "ruff" },
+]
+
[package.metadata]
requires-dist = [
{ name = "arrow", specifier = "==1.4.0" },
@@ -160,6 +165,9 @@ requires-dist = [
{ name = "waitress", specifier = "==3.0.2" },
]
+[package.metadata.requires-dev]
+dev = [{ name = "ruff", specifier = "==0.16.0" }]
+
[[package]]
name = "itsdangerous"
version = "2.2.0"
@@ -423,6 +431,31 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/81/c4/34e93fe5f5429d7570ec1fa436f1986fb1f00c3e0f43a589fe2bbcd22c3f/pytz-2025.2-py2.py3-none-any.whl", hash = "sha256:5ddf76296dd8c44c26eb8f4b6f35488f3ccbf6fbbd7adee0b7262d43f0ec2f00", size = 509225, upload-time = "2025-03-25T02:24:58.468Z" },
]
+[[package]]
+name = "ruff"
+version = "0.16.0"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/4d/94/1e5e4967626faf12fa56999cd6222dff6992ceb086ad7945756baf70c7a7/ruff-0.16.0.tar.gz", hash = "sha256:e460aafd5495ec89efaa6ced2e4a9a581116451e1c88b9d37ef497e0f8e93982", size = 4790557, upload-time = "2026-07-23T19:11:30.981Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/4b/81/1c8818fee7ce1a04cd7d1b3172e0a8f8e4f1dc4feb7fc390e16daa8af323/ruff-0.16.0-py3-none-linux_armv6l.whl", hash = "sha256:e5115729eb08c585e5121978ba5d5b60caeae394ce21b9fb5e6cd33a1c6c9b1e", size = 10754633, upload-time = "2026-07-23T19:10:46.415Z" },
+ { url = "https://files.pythonhosted.org/packages/23/df/beaf59c09d68db84304d555f188b276a77132a5d5b0b67a5c762aa143628/ruff-0.16.0-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:3c954b1d580bfa035b41654f7858cc7e71d5fc3ac5b723dd62bd9133830ed522", size = 10969164, upload-time = "2026-07-23T19:10:50.271Z" },
+ { url = "https://files.pythonhosted.org/packages/42/ce/741cd197496a1abbf51352710fd15ed995d2a2be87189c1da26a450d6e83/ruff-0.16.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:e01c21d10eb1b29f47b7454e1f4056db9a3f0260c646aa88457c610291db9f81", size = 10488846, upload-time = "2026-07-23T19:10:52.639Z" },
+ { url = "https://files.pythonhosted.org/packages/52/2a/a2db8e88cade358f5cdcb05674a917751074109315d014eb6352d9a893f7/ruff-0.16.0-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:6e364e5ed22ed8dc05082fd78e35308618260907ac2d3c1d637b2e682415b6c9", size = 10889729, upload-time = "2026-07-23T19:10:54.89Z" },
+ { url = "https://files.pythonhosted.org/packages/42/65/62a771694ebd63029dc953e27dbad40e1588bd4860ff9fe881018fddaa49/ruff-0.16.0-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:d327b8fc113a1d4421a04f3839d3752057c8dd1ee320223a6f3f52d04ada462a", size = 10568275, upload-time = "2026-07-23T19:10:56.993Z" },
+ { url = "https://files.pythonhosted.org/packages/3f/e2/ced249fe8af5f086c5c58cc21cc3356d50f32f7401c5df87050c999620a7/ruff-0.16.0-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:a9b50c55e263103586b3dcf5f73d479eb8cb5fdb6098fec59a62891dab653717", size = 11385112, upload-time = "2026-07-23T19:10:59.615Z" },
+ { url = "https://files.pythonhosted.org/packages/87/0b/05154977a8fd69eeb6c103271f55403bfd8711f5c0f8ed07489d95a504e7/ruff-0.16.0-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:0ff4a79ce3ec0172f3241943835de1c4cb4e2dcd07f0f8c2d02603dbbbee4b17", size = 12207008, upload-time = "2026-07-23T19:11:02.154Z" },
+ { url = "https://files.pythonhosted.org/packages/fb/29/98225831a3a1eab0e02f4acc6ca6559a98611dcc68b6965ff4b7234627c1/ruff-0.16.0-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:e95c448fca1fb2a18372a9440926c5a6ee789639bb975c72e7ae6d0b04218ab4", size = 11650842, upload-time = "2026-07-23T19:11:04.557Z" },
+ { url = "https://files.pythonhosted.org/packages/91/66/6bd3cf90500653d55dc0ffc8507aa8300bd49d0214b2e8cb4d3fef2943ba/ruff-0.16.0-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:4f11a8d11010301d0a398a2fdef67691feca7294da6aef55e2150e8fa2cd520b", size = 11400718, upload-time = "2026-07-23T19:11:09.233Z" },
+ { url = "https://files.pythonhosted.org/packages/8e/a2/a54eb4eae05d66364050a5d3b8a9c5ef88196531b3cbe7109d873f87f819/ruff-0.16.0-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:48044c678e9cb8698246c99b14aaccfa6601dea7379eb48a6f8f73f7a6d86cd0", size = 11426177, upload-time = "2026-07-23T19:11:11.994Z" },
+ { url = "https://files.pythonhosted.org/packages/1a/be/16e3eea4b2a478a496919f5e36f17c4559e54620bd3bbac5d6affa068006/ruff-0.16.0-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:7aa0959bad8eb8bef50340154fc9b58678dae31fa4293afa38b44b6e552c0213", size = 10856126, upload-time = "2026-07-23T19:11:14.221Z" },
+ { url = "https://files.pythonhosted.org/packages/a2/84/252eb8b868a16eec7257c14f504f77537e734b2d69c762e639e588e304a3/ruff-0.16.0-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:28ea2b7df8ebf7f9da6b7d47b230ab48f387c0a29be3b474c4d0740e197bb9af", size = 10571208, upload-time = "2026-07-23T19:11:16.378Z" },
+ { url = "https://files.pythonhosted.org/packages/21/09/817a482f542f7570cbb4554b26e896610c7114f539b1d9e2d2145bf6bef6/ruff-0.16.0-py3-none-musllinux_1_2_i686.whl", hash = "sha256:33a3dfac8c35f81498dea9181bccc2f4c4bc8f1521a1dd9406e77643e0f0fb09", size = 11063329, upload-time = "2026-07-23T19:11:19.173Z" },
+ { url = "https://files.pythonhosted.org/packages/2e/23/9403c180ca1cb9b1f7335f5c3e5305c09d49ea5b345196682a36028bde4a/ruff-0.16.0-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:a5237a0bda500d30d81b8e07a6973a5cbc772864cbf746ae2f4e8a2e01c9f4ed", size = 11489751, upload-time = "2026-07-23T19:11:21.74Z" },
+ { url = "https://files.pythonhosted.org/packages/b2/1d/1b2ef7bcde851c78d7f17f1cca13fd6dc695fc4b3d6197941e72cae5b132/ruff-0.16.0-py3-none-win32.whl", hash = "sha256:7fab76fa065c873f41ff744347c6e77bcc3dfec4bcc754dc26b63d23c0f7f5fb", size = 10785885, upload-time = "2026-07-23T19:11:23.947Z" },
+ { url = "https://files.pythonhosted.org/packages/b2/a3/d5e4ef7a56be3f928ffb90b94c25ba7d3cb9c7fe0736aeaaedf361770712/ruff-0.16.0-py3-none-win_amd64.whl", hash = "sha256:429c117f022bf481fabd9d551e7a3952b24c65e6ef44337ea09d90bebef14472", size = 11923141, upload-time = "2026-07-23T19:11:26.409Z" },
+ { url = "https://files.pythonhosted.org/packages/cb/9a/8415f2657cbe200f41a4531ccededf135505a92d4a012229121f885b26f9/ruff-0.16.0-py3-none-win_arm64.whl", hash = "sha256:14296fedcd2705c77ab8235439278bbb38f285cf7da5528b00b3e330c3d4872d", size = 11273407, upload-time = "2026-07-23T19:11:28.705Z" },
+]
+
[[package]]
name = "setuptools"
version = "82.0.1"
diff --git a/frontend/src/components/PanelIssueNotice.jsx b/frontend/src/components/PanelIssueNotice.jsx
new file mode 100644
index 0000000..db0e68b
--- /dev/null
+++ b/frontend/src/components/PanelIssueNotice.jsx
@@ -0,0 +1,35 @@
+const ISSUE_STYLES = {
+ error: {
+ container: "border-red-500/60 bg-red-950/55",
+ label: "text-red-300",
+ message: "text-red-100",
+ },
+ warning: {
+ container: "border-yellow-500/60 bg-yellow-950/45",
+ label: "text-yellow-300",
+ message: "text-yellow-100",
+ },
+};
+
+export default function PanelIssueNotice({ type, children }) {
+ const styles = ISSUE_STYLES[type];
+ if (!styles || children == null || children === "") return null;
+
+ return (
+
+ A red Unknown Events marker + appears when metadata or snapshot data could not be read. Snapshots + that simply fall outside the selected range are not flagged. It + indicates that one or more events occurred in that part of the + timeline, even when the exact events cannot be determined. The next + readable event is compared with the previous readable metadata. Its + details show the metadata changes, and when the snapshot changed they + also show that snapshot's operation. +
@@ -400,6 +410,10 @@ const SECTIONS = [
{search
diff --git a/frontend/src/pages/GraphPage.jsx b/frontend/src/pages/GraphPage.jsx
index f094f6d..b6e47a7 100644
--- a/frontend/src/pages/GraphPage.jsx
+++ b/frontend/src/pages/GraphPage.jsx
@@ -7,6 +7,7 @@ import {
PanelHeader,
PANEL_STATUS_BADGE_CLASS,
} from "../components/PanelContent";
+import PanelIssueNotice from "../components/PanelIssueNotice";
import {
UI_DIALOG_SECTION_TITLE_CLASS,
UI_POPUP_HINT_CLASS,
@@ -683,7 +684,10 @@ export default function GraphPage() {
const sticky = stickyNode
? {
rows: Object.entries(stickyNode.details)
- .filter(([label]) => label.toLowerCase() !== "type")
+ .filter(
+ ([label]) =>
+ !["type", "error", "warning"].includes(label.toLowerCase()),
+ )
.map(([label, value]) => ({
label,
value,
@@ -832,6 +836,12 @@ export default function GraphPage() {
{isInspectMode && (
🔒 Locked View
)}
+