---
pipeline:
  name: mti300ahrs_location_group_and_restructure
transform:
  image: us-central1-docker.pkg.dev/neon-shared-service/neonscience/neon-is-loc-grp-strc-comb:v1.1.0
  cmd:
  - sh
  - "-c"
  - |-
    /bin/bash <<'EOF'
    # Use bash-scrict mode. See http://redsymbol.net/articles/unofficial-bash-strict-mode/
    set -euo pipefail
    IFS=$'\n\t'
    # Refresh interim directories with each datum (otherwise they persist and cause probs)
    rm -rf /tmp/pfs/filter_joined
    rm -rf /tmp/pfs/structured
    rm -rf /tmp/pfs/structuredCopy
    mkdir -p /tmp/pfs/filter_joined
    # Run first module - filter-joiner (using environment variables below as input parameters)
    python3 -m filter_joiner.filter_joiner_main
    # Run second module - structure repo by location
    Rscript ./flow.loc.repo.strc.R \
      DirIn=/tmp/pfs/filter_joined \
      DirOut=/tmp/pfs/structured \
      DirErr=/pfs/out/errored_datums \
      Comb=TRUE
    # Copy output to another interim folder to destroy links (cannot daisy chain links from pfs input to output)
    cp -rL /tmp/pfs/structured /tmp/pfs/structuredCopy  || : # Allow to fail without exit code (happens if step above produced no output)
    rm -rf /tmp/pfs/filter_joined 
    rm -rf /tmp/pfs/structured 
    # Run third module - merge data by location
    Rscript ./flow.loc.data.trnc.comb.R \
      DirIn=/tmp/pfs/structuredCopy \
      DirOut=/pfs/out \
      DirErr=/pfs/out/errored_datums \
      "DirSubCombData=data|flags" \
      DirSubCopy=location
    EOF
  env:
    # Environment variables for filter-joiner
    CONFIG: |
      ---
      # In Pachyderm root will be index 0, 'pfs' index 1, and the repo name index 2.
      # Metadata indices will typically begin at index 3.
      input_paths:
        - path:
            name: DATA_PATH
            # Filter for data directory
            glob_pattern: /pfs/DATA_PATH/mti300ahrs/*/*/*/*/**
            # Join on named location (already joined below by day)
            join_indices: [7]
            outer_join: true
        - path:
            name: LOCATION_PATH
            # Filter for data directory
            glob_pattern: /pfs/LOCATION_PATH/mti300ahrs/*/*/*/*/**
            # Join on named location (already joined below by day)
            join_indices: [7]
    OUT_PATH: /tmp/pfs/filter_joined
    LOG_LEVEL: INFO
    RELATIVE_PATH_INDEX: "3"
    LINK_TYPE: COPY # options are COPY or SYMLINK. Use COPY for combined module.
    # Environment variables for R modules
    PARALLELIZATION_INTERNAL: '1' 
input:
  join:
  - pfs:
      name: DATA_PATH
      repo: mti300ahrs_data_source_trino
      glob: /mti300ahrs/(*)/(*)/(*)
      joinOn: $1/$2/$3
      outer_join: true
      empty_files: false # Make sure to use false if LINK_TYPE=COPY. Can also be set to false for LINK_TYPE=SYMLINK.
  - pfs:
      name: LOCATION_PATH
      repo: mti300ahrs_location_asset_assignment
      glob: /mti300ahrs/(*)/(*)/(*)
      joinOn: $1/$2/$3
      empty_files: false # Make sure to use false if LINK_TYPE=COPY. Can also be set to false for LINK_TYPE=SYMLINK.
parallelism_spec:
  constant: 1
autoscaling: true
resource_requests:
  memory: 3G
  cpu: 5.5
resource_limits:
  memory: 6G
  cpu: 7
sidecar_resource_requests:
  memory: 3G
  cpu: 0.5
sidecar_resource_limits:
  memory: 8Gi
  cpu: 1.2
datum_set_spec:
  number: 1
scheduling_spec:
  node_selector:
    cloud.google.com/gke-ephemeral-storage-local-ssd: "true"
    nodepool.neonscience.org/pipeline: "yes"
    cloud.google.com/gke-spot: "true"
pod_spec: |-
  { "tolerations": [
    {
      "key": "nodepool.neonscience.org/pipeline",
      "operator": "Exists"
    },
    {
      "effect": "NoSchedule",
      "key": "cloud.google.com/gke-spot",
      "operator": "Exists"
    }  
  ] }
