Repository navigation
Ray upload - #264
Merged
Merged
Ray upload#264
Conversation
…n first use Tables for the job record, stages, the Cloud SQL import (intent row before the request), CSV/parquet parts and spatial chunks, all cascading from upload_jobs for the 30-day retention sweep. get_engine() creates the database and schema if missing (advisory-locked, versioned); the store gives record merges that a cancel blocks, a lease with epoch fencing, insert-once plans and batched chunk upserts that never undo COMPLETED. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…dices, as plain functions process_sub_batch_points is the body of process_and_insert_sub_batch, which now calls it. process_chunk_points runs a whole chunk in-process (for the Ray upload's chunk tasks). rebuild_segmentation_indices runs the index rebuild that rebuild_indices_for_spatial_lookup only builds as a chain. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
… adoption and 409 backoff - upload_state.status_sink: in a StoreSink context, update_job_status, get_job_status, is_upload_cancelled and the import state helpers use the upload state database (fenced on the driver's lease epoch); a part runner can replace _run_csv_part_tasks. Without either, everything is as before (redis, celery). - process_csv takes a fixed timestamp, so a resumed run writes the same output. - The import is recorded as requested before it is POSTed; a later attempt that finds it requested but not recorded adopts the matching Cloud SQL operation (find_import_operation) instead of importing twice. import_csv waits while the instance is busy (HTTP 409) for up to an hour. - GCS compose is safe to repeat after it composed but did not finish deleting parts. - wait_for_backend_work generalises wait_for_index_builds (e.g. CLUSTER too). - drop_upload_tables works on staging databases without combined_table_metadata. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…ily cleanup - rayjobs/entrypoints/bulk_upload.py: runs an upload's stages in order (the celery tasks' own bodies, reporting through a StoreSink), resuming from the upload state database after a crash: finished stages are skipped, only unfinished CSV/parquet parts and spatial chunks run again. A lease with an epoch keeps a second driver from writing; failures are permanent (error, staging discarded) after MAX_STAGE_ATTEMPTS or for errors a retry would repeat, otherwise exit 1 for KubeRay to retry. A cancel stops before the transfer; once the transfer is entered a cancel is refused (atomically, store.enter_stage/request_cancel). - rayjobs/upload_tasks.py: the worker side, in a config-only Flask app. - rayjobs/runner.py: RayRunner and InlineRunner behind one interface, and run_bounded (bounded in-flight, per-item retry decisions, stop between waits). - submitter: ray_upload_enabled, a per-job activeDeadlineSeconds, DNS-safe names and label values, Retrying is not finished, delete tolerates 404, list_rayjobs. - upload_state/cleanup.py: 30-day purge, ends jobs whose RayJob is over, drops leftover staging of failed jobs, deletes old upload RayJob CRs. - store: chunk upserts take one row per chunk. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…N for the pages - With ray.upload on, /api/process/start creates the job in the upload state database (409 if the table already has a running upload) with everything the driver needs, including the frozen time stamps, and submits its RayJob. - Status, user-jobs, the admin uploads list and the spatial lookup endpoints (status, active, all-active-direct, workflow-details) read the upload state database first (StoreCheckpointView mirrors RedisCheckpointManager's readers; the more recently updated run of a workflow wins), redis otherwise. - A job whose RayJob ended or vanished while it was running is marked error (cancelled, if a cancel was pending) instead of staying pending; a retrying RayJob shows in the phase. Kubernetes is asked at most every 10 s per job. - Cancel is a request the driver acts on; refused once the transfer has started. - The app creates the upload state database at startup when Ray uploads are on. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…setup early - run_task re-raises an exception Ray cannot pickle (google.api_core's hold a lock) as TaskFailed with its type and message; the driver saw only "isn't serializable" for ltv7's empty BIGTABLE_INSTANCE. - SYSTEMIC_FAILURE_CHUNKS (20) different chunks failing with the same error and none succeeding stops the spatial chunks stage, instead of trying every chunk 10 times (the missing graphene token and the empty Bigtable instance, both setup errors). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
…NTENANCE_WORK_MEM) An upload builds its staging annotation indexes and, in the transfer, its production indexes one at a time, so a large value (6GB on minnie) is held by one session per upload and only for the build (SET LOCAL). Other index builds keep add_index's 1GB: the materialization workflows build many at once. add_index now uses SET LOCAL instead of SET ... then SET 64MB, which left the pooled connection at 64MB, below the instance's 2GB, for later maintenance commands. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
adding uploading via ray jobs