I'm going to dump a couple thoughts and questions here for grooming purposes (we can discuss more on Tuesday):

The naive way to find the right timestamp without firing up qlever is to iterate over the files and keep track of the oldest timestamp seen. I have a small script to do this, which merely iterates over all the rows in the glob of *.nt files and parses timestamps with a regex and keeps track of the oldest one, but it takes 47 minutes to run over all 1024 shards. (I'm sure it can be improved!) Assuming we want to do something like this, I think it will need to happen in a separate KubernetesPodOperator, I think ideally after the indexing, so that the manifest can be written at the same time (the manifest and the timestamp ultimately end up in the same .json file in the data to be synced to S3).

Syncing to S3 using rclone in a KubernetesPodOperator worked well in the data-dump DAG; we should do that again.

FYI for setting up the KPO command: I had an issue with too many open files when running the indexer over all of main; see my note about ulimit here: T427030#12011716

Question: What's the best way to interact with ephemeral storage? Should we request a PVC before doing the copy/index/sync and then delete it afterwards, or is there some other good way to interact with the underlying pod storage?

The naive way to find the right timestamp without firing up qlever is to iterate over the files and keep track of the oldest timestamp seen. I have a small script to do this, which merely iterates over all the rows in the glob of *.nt files and parses timestamps with a regex and keeps track of the oldest one, but it takes 47 minutes to run over all 1024 shards.

If we want to do this outside of the indexer POD, it's better to do a simple spark job / sql query on the parquet files we generate prior
to the n3 export.

Something like

> SELECT MAX(from_iso8601_timestamp(regexp_extract(object, '"([^"]*)"', 1))) AS dateModified
FROM wikibase_rdf_subgraphs
WHERE snapshot = '20260525'
    AND scope = 'wikidata_main'
    AND predicate = '<http://schema.org/dateModified>'

dateModified
2026-05-27 06:30:03.000 UTC

runs in about 20seconds on presto. I filtered by scope = 'wikidata_main' to reduce the search space. The timstamp is the same for both graphs.

Syncing to S3 using rclone in a KubernetesPodOperator worked well in the data-dump DAG; we should do that again.

This will run from the indexing pod, right? Do we need to bundle rclone with the qlever docker image?

Question: What's the best way to interact with ephemeral storage? Should we request a PVC before doing the copy/index/sync and then delete it afterwards, or is there some other good way to interact with the underlying pod storage?

There's a thread in slack https://wikimedia.slack.com/archives/C0B8DMX1ZGS/p1781117180927639?thread_ts=1780909462.306089&cid=C0B8DMX1ZGS

We don't need a PVC because the data lifecycle within the pod is ephemeral and data is persisted only on S3. emptyDir seems a natural fit, but we need to size request/limit accordingly.

This will run from the indexing pod, right? Do we need to bundle rclone with the qlever docker image?

Ah. I was originally thinking this would be a sequence of tasks: one to rclone to a PVC (one that would get cleaned up, not to store data across runs of the DAG), one to actually do the indexing using the Qlever image, and one to do rclone again. We could have a PVC that we create and delete for this specific series of tasks. I didn't appreciate until now that you wanted to do this in a single pod, but that would be cool if possible (I am biased because the only way I was actually able to interact with S3 was via a KubernetesPodOperator that did rclone, so I feel like anything else will be harder than we expect.) Thanks for that slack pointer, somehow I had missed that convo. I am interested to see if this will work!

Ah. I was originally thinking this would be a sequence of tasks: one to rclone to a PVC (one that would get cleaned up, not to store data across runs of the DAG), one to actually do the indexing using the Qlever image, and one to do rclone again. We could have a PVC that we create and delete for this specific series of tasks.

I would rather not having to deal with managing a pvc lifecyle and state for this, since data stored locally is ephemeral.
We could hydrate a POD emptyDir in the same way we do with the init container. Reposting here for visiblity,
but this is the workflow I had in mind:

We plan to schedule the qlever indexing job from airflow to k8s. This will be an ephemeral pod that runs the qlever-index binary (we bundle it with the qlever docker image). In my mental model, the lifecyle should be rouhgly:

1. We persist  n3 files (for the graphs) in S3. Same spark job that Search uses in import_ttl. See T429253: data-reload: implement export_n3 task.
2. We launch the qlever indexer job from a KubernetesPodOperator operator. Then, inside the POD:
3. We copy data from S3 to local storage
4. We build the qlever index on local storage
5. We persist the index (and a manifest) back into an S3 bucket
6. The pod goes down, and the local volume can be wiped.

Open question: should we create a dedicated namespace for this, or reuse WDQS? We need to avoid resource contention in the namespace between qlever-indexer and qlever-server PODs.

gmodena updated https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2351

Draft: wikidata: index RDF graph with qlever

Change #1309261 had a related patch set uploaded (by Btullis; author: Btullis):

[operations/deployment-charts@master] admin_ng: configure topolvm storage classes and node placement

https://gerrit.wikimedia.org/r/1309261

Change #1310066 had a related patch set uploaded (by Gmodena; author: Gmodena):

[operations/deployment-charts@master] wdqs: load qlever index from qlever-index/ S3 prefix

https://gerrit.wikimedia.org/r/1310066

Change #1309261 merged by jenkins-bot:

[operations/deployment-charts@master] admin_ng: configure topolvm storage classes and node placement

https://gerrit.wikimedia.org/r/1309261

Change #1310066 merged by jenkins-bot:

[operations/deployment-charts@master] wdqs: load qlever index from qlever-index/ S3 prefix

https://gerrit.wikimedia.org/r/1310066

Apologies for the delay in getting back to you about this. I'll see if I can help to make things clearer and less confusing.
The good news is that now we have TopoLVM in production and I think it will make things much easier, whichever direction we take.

In T428235#12023553, @gmodena wrote:

Ah. I was originally thinking this would be a sequence of tasks: one to rclone to a PVC (one that would get cleaned up, not to store data across runs of the DAG), one to actually do the indexing using the Qlever image, and one to do rclone again. We could have a PVC that we create and delete for this specific series of tasks.

I would rather not having to deal with managing a pvc lifecyle and state for this, since data stored locally is ephemeral.

This is the fundamental question to be answered:
Is it one pod, or many tasks?

@lerickson has been planning for this to be a sequence of tasks, each performing a small step, potentially including the creation and deletion of a volume.
@gmodena has been working on the basis that the life-cycle of the volume would not be managed by the DAG, but would just be available permanently, pre-allocated.

If everything that needs to happen (download,index,upload) can be called from a single script, then we already have an easy to make everything happen in a single pod.
This lifecycle of having an ephemeral volume that only needs to last the lifetime of a single pod is already proven (See: T428895#12115579)

Before we had TopoLVM working, @gmodena proposed this change: https://gerrit.wikimedia.org/r/c/operations/deployment-charts/+/1305852 to pre-allocate the volumes.
There are a few problems with the way that this was intended to work, not least requiring root to deploy, and pre-allocate the volumes, as well as perform housekeeping on the volume contents.

I feel that what we need is:

• a very simple method to create a PVC that lasts only the duration of a DAG run
• ...but does not come with a heavy burden of lifecycle management

This would allow us to support what @lerickson was hoping for, where each individual task is as simple as possible (download,index,upload).
...while also supporting what @gmodena wanted, which was not having to care much about lifecycle management and state for a temporary resource.

So here is what I propose:

• The Data-Platform-SRE team creates factory task templates to support creating and deleting a scratch PVC.
• We also create an integration test DAG on the test-k8s instance that a) shows how to use them and b) proves that they work
• Your DAG then becomes create >> download >> index >> upload >> delete using separate tasks. The create and delete tasks would be very lightweight.

The benefits of this method are:

• You never have to deal with the question of: "what is already on the volume?" because the volume is freshly created per DAG run. No state management.
• You can name the PVC per run (qlever-index-{{ run_id }}) and set max_active_runs=1 on the DAG, to avoid clashing on resource names and requesting too manay volumes at once.
• Your indexing DAG is not actually tied to a single host.
  • Any host that has sufficient capacity in the topolvm-balanced (or whatever) storageclass can be selected for the indexing task.
  • This would make a big difference in terms of availability during kubernetes cluster restarts, or other maintenance operations.
• If a certain task fails, then the DAG would not have to go back to the beginning.
  • i.e. if an index tasks takes 5 hours, but the upload task failed, the PVC containing the new index would still exist, so only the upload and delete task would remain.
• These factory task templates would be reusable for other teams who need storage (e.g. ML and SRE)

Change #1311846 had a related patch set uploaded (by Btullis; author: Btullis):

[operations/deployment-charts@master] airflow: let DAGs manage run-scoped scratch volumes on TopoLVM

https://gerrit.wikimedia.org/r/1311846

btullis opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2404

Add a factory for run-scoped scratch volumes on node-local storage

Change #1311846 merged by jenkins-bot:

[operations/deployment-charts@master] airflow: let DAGs manage run-scoped scratch volumes on TopoLVM

https://gerrit.wikimedia.org/r/1311846

btullis merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2404

Add a factory for temporary volumes scoped to a DAG run

brouberol opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2417

Define a way to render jinja templates. Use it to render the temporary PVC

brouberol merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2417

Define a way to render jinja templates. Use it to render the temporary PVC

Hey @BTullis,

Thanks for tihs!

In T428235#12131966, @BTullis wrote:

So here is what I propose:

• The Data-Platform-SRE team creates factory task templates to support creating and deleting a scratch PVC.
• We also create an integration test DAG on the test-k8s instance that a) shows how to use them and b) proves that they work
• Your DAG then becomes create >> download >> index >> upload >> delete using separate tasks. The create and delete tasks would be very lightweight.

What is the timeline for this?

The current approach https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2351 was agreed upon put in place to allow us to iterate fast and have a working version of qlever indexing e2e, while topolvm work was ongoing. But it has been stalled for a while now.

I am not particularly invested in one solution or another, but I would really to not have to worry about the DAG managing state (files on disk) and POD allocation.

create >> download >> index >> upload >> delete

Modulo managing pvc creation/deletion this is the workflow we have in draft.

With toplovlm in place, do we need higher level factory classes or can we start implementing a solution (and then refactor, if factories make sense)? What are the moving parts currently missing for refactoring !2351 ? Can we just pull in temp_volume

volume = temp_volume(size="1Gi", location="node-local", medium="ssd")

And rewrite the chain?

Before DAGs can use this, the instance namespace needs an extra_rbac rule in operations/deployment-charts allowing the airflow service account to manage PVCs, and it must be listed in the tenantNamespaces of the release providing the storage (topolvm for node-local volumes, ceph-csi-rbd or ceph-csi-cephfs for network ones).

This is not in place yet on the airflow_wikidata instance, right?

In T428235#12149901, @gmodena wrote:

Hey @BTullis,

• Your DAG then becomes create >> download >> index >> upload >> delete using separate tasks. The create and delete tasks would be very lightweight.

What is the timeline for this?

This is ready now.

Here is the reference implementation in action: https://airflow-test-k8s.wikimedia.org/dags/setup_temporary_local_volume
Here is the code for that reference implementation: test_k8s/dags/temp_volume_dag.py
You can use this code in your DAG now.

If you want to see how the task templates work, they are defined here.

...I would really to not have to worry about the DAG managing state (files on disk) and POD allocation.

Yes, I feel that is what we have delivered with this solution. If you have any queries or concerns about it, please don't hesitate to let us know.

create >> download >> index >> upload >> delete

Modulo managing pvc creation/deletion this is the workflow we have in draft.

Yes, I feel that your draft can now be adapted to use this workflow and start moving again, immediately.
The README in that draft says:

Under the hood we use a PV with node affinity to two high mem/storage hosts in the k8s cluster.
Related infra changes are implemented in 1305854

You will now be able to remove those two statements because with this temporary volume mechanism, you no longer need any static definition of PVCs or other infrastructure, outside of the DAG.

With toplovlm in place, do we need higher level factory classes or can we start implementing a solution (and then refactor, if factories make sense)? What are the moving parts currently missing for refactoring !2351 ? Can we just pull in temp_volume

volume = temp_volume(size="1Gi", location="node-local", medium="ssd")

And rewrite the chain?

Yes, pretty much exactly that. Just refer to the example DAG here.

Before DAGs can use this, the instance namespace needs an extra_rbac rule in operations/deployment-charts allowing the airflow service account to manage PVCs, and it must be listed in the tenantNamespaces of the release providing the storage (topolvm for node-local volumes, ceph-csi-rbd or ceph-csi-cephfs for network ones).

This is not in place yet on the airflow_wikidata instance, right?

Yes, this is already in place for the airflow-wikidata instance.

See these references:

• https://github.com/wikimedia/operations-deployment-charts/blob/master/helmfile.d/dse-k8s-services/airflow-wikidata/values-production.yaml#L5-L11
• https://github.com/wikimedia/operations-deployment-charts/blob/master/helmfile.d/admin_ng/values/dse-k8s-eqiad/topolvm-values.yaml#L9

The one thing that we haven't done it to enable it for the airflow-dev namespace, so it will not yet work with the airflow-devenv environment.
We wanted to give some thought to quotas etc before allowing anyone in any devenv to create volumes wherever they want.

trueg opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2448

wikidata: index RDF graph with qlever

trueg closed https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2351

Draft: wikidata: index RDF graph with qlever

trueg opened https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-qlever/-/merge_requests/13

New configuration parameter QLEVER_INDEX_DIR

trueg updated https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2450

wikidata data-reload: Add Qlever index creation

trueg merged https://gitlab.wikimedia.org/repos/wikidata-platform/wdqs/wdqs-qlever/-/merge_requests/13

New configuration parameters QLEVER_INDEX_DIR and QLEVER_INDEXER_CFG_RAW

trueg closed https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2448

wikidata: index RDF graph with qlever

lerickson merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2450

wikidata data-reload: Add Qlever index creation

lerickson opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2459

wikidata data-reload: Change WMFKubernetesPodOperator to KubernetesPodOperator

trueg merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2459

wikidata data-reload: Change WMFKubernetesPodOperator to KubernetesPodOperator

lerickson updated https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2460

wikidata data-reload: use preexisting kubernetes pod operator base args that work in other tasks

lerickson merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2460

wikidata data-reload: use preexisting kubernetes pod operator base args that work in other tasks

lerickson opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2461

wikidata data-reload: update security context for KPO containers.

lerickson merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2461

wikidata data-reload: update security context for KPO containers.

lerickson opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2463

wikidata data-reload: Add tolerations so that the pods can be scheduled on WDQS nodes.

lerickson merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2463

wikidata data-reload: Add tolerations so that the pods can be scheduled on WDQS nodes.

lerickson opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2481

wikidata data-reload: Update qlever image

lerickson merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2481

wikidata data-reload: Update qlever image to include a bash fix

lerickson opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2482

wikidata data-reload: log the JSON to write, and (temporarily) stop deleting temp files

trueg merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/2482

wikidata data-reload: log the JSON to write, and (temporarily) stop deleting temp files

Moving to "waiting" as work seems to be happening in Wikidata Platform Team. Ping us if you need our help.

Marked for grooming so we can rescore the task and align on the correct scope for the upcoming sprint

I do have some questions for SRE in this Slack thread, but I will also post them here for better visibility!

We are having issues with the temp volume. I suspect it's because we are using it from 2 pods simultaneously, which seems to not be allowed. Given that SRE implemented the temp volume in the first place, I would like to check whether there's any issue with using two volumes at the same time in a single DAG? (Provided space exists, of course. If this is an easy "no, no problem!", that's great! Just wanted a quick check.)

The specific error message I saw on a DAG run that failed in one indexing step (but succeeded in the other, suggesting that the concurrency might have been the issue) was:

'message': '0/31 nodes are available: 1 ' 'Insufficient cpu, 2 node(s) had ' 'untolerated taint ' '{node-role.kubernetes.io/control-plane: ' '}, 21 Insufficient memory, 8 node(s) ' 'had volume node affinity conflict. ' 'preemption: 0/31 nodes are available: ' '10 Preemption is not helpful for ' 'scheduling, 21 No preemption victims ' 'found for incoming pod.'

Question: would this message be consistent with us attempting to use a temp volume by 2 pods at the same time, or does this point to yet another issue on top of the forbidden concurrency?

Finally, one more question:

• is there a tech reason for no concurrency for node-local volumes, or would this be straightforward to also allow for us?