pipedag.yaml

name: pipedag_tests
table_store_connections:
  postgres:
    args:
      url: "postgresql://sa:Pydiverse23@127.0.0.1:6543/{instance_id}"
      create_database_if_not_exists: true

  mssql:
    args:
      url: "mssql+pyodbc://sa:PydiQuant27@127.0.0.1:1433/{instance_id}?driver=ODBC+Driver+18+for+SQL+Server&encrypt=no"
      create_database_if_not_exists: true
      strict_materialization_details: false

  ibm_db2:
    args:
      url: "db2+ibm_db://db2inst1:password@localhost:50000/testdb"
      schema_prefix: "{instance_id}_"
      create_database_if_not_exists: false

  duckdb:
    args:
      url: "duckdb:////tmp/pipedag/{instance_id}/db.duckdb"
      create_database_if_not_exists: true

  snowflake:
    args:
      url: "snowflake://{$SNOWFLAKE_USER}:{$SNOWFLAKE_PASSWORD}@{$SNOWFLAKE_ACCOUNT}/pipedag/DBO?warehouse=pipedag&role=accountadmin"
      schema_prefix: "{instance_id}_"
      create_database_if_not_exists: true

blob_store_connections:
  file:
    class: "pydiverse.pipedag.backend.blob.FileBlobStore"
    args:
      base_path: "/tmp/pipedag/blobs"

  no_blob:
    class: "pydiverse.pipedag.backend.blob.NoBlobStore"
    args:

instances:
  __any__:
    network_interface: "127.0.0.1"
    auto_table:
      - "pandas.DataFrame"
      - "sqlalchemy.sql.expression.TextClause"
      - "sqlalchemy.sql.expression.Selectable"

    # Attention: For disable_kroki: false, stage and task names might be sent to the kroki_url.
    #   You can self-host kroki if you like:
    #   https://docs.kroki.io/kroki/setup/install/
    disable_kroki: true
    kroki_url: "https://kroki.io"

    fail_fast: true
    instance_id: pipedag_default
    table_store:
      table_store_connection: postgres
      class: "pydiverse.pipedag.backend.table.SQLTableStore"
      args:
        print_materialize: true
        print_sql: true
      hook_args:
        pandas:
          dtype_backend: "arrow"

    blob_store:
      blob_store_connection: file

    lock_manager:
      class: "pydiverse.pipedag.backend.lock.DatabaseLockManager"

    orchestration:
      class: "pydiverse.pipedag.engine.SequentialEngine"

  # Instances used by pytest

  ## Database Instances

  postgres:
    instance_id: pd_postgres
    table_store:
      table_store_connection: postgres

  postgres_unlogged:
    instance_id: pd_postgres_unlogged
    table_store:
      table_store_connection: postgres
      args:
        materialization_details:
          __any__:
            unlogged: true
      hook_args:
        pandas:
          dtype_backend: "numpy"  # also test numpy backed pandas
    blob_store:
      blob_store_connection: no_blob

  mssql:
    instance_id: pd_mssql
    table_store:
      table_store_connection: mssql
      args:
        disable_pytsql: true
    blob_store:
      blob_store_connection: no_blob

  mssql_pytsql:
    instance_id: pd_mssql_pytsql
    table_store:
      table_store_connection: mssql
      args:
        disable_pytsql: false
        pytsql_isolate_top_level_statements: true
    blob_store:
      blob_store_connection: no_blob

  ibm_db2:
    instance_id: pd_ibm_db2
    stage_commit_technique: READ_VIEWS
    table_store:
      table_store_connection: ibm_db2
    blob_store:
      blob_store_connection: no_blob

  ibm_db2_avoid_schema:
    instance_id: pd_ibm_db2_avoid_schema
    stage_commit_technique: READ_VIEWS
    table_store:
      table_store_connection: ibm_db2
      args:
        avoid_drop_create_schema: true
    blob_store:
      blob_store_connection: no_blob

  ibm_db2_materialization_details:
    instance_id: pd_ibm_db2_materialization_details
    stage_commit_technique: READ_VIEWS
    table_store:
      table_store_connection: ibm_db2
      args:
        strict_materialization_details: false
        default_materialization_details: "no_compression"
        materialization_details:
          __any__:
            compression: [ "COMPRESS YES ADAPTIVE" ]
          no_compression:
            compression: ""
          value_compression:
            compression: "VALUE COMPRESSION"
          static_compression:
            compression: "COMPRESS YES STATIC"
          adaptive_value_compression:
            compression: [ "COMPRESS YES ADAPTIVE", "VALUE COMPRESSION" ]
          table_space:
            table_space_data: "S1"
            table_space_index: "S2"
            table_space_long: "S3"
    blob_store:
      blob_store_connection: no_blob


  duckdb:
    instance_id: pd_duckdb
    stage_commit_technique: READ_VIEWS
    table_store:
      table_store_connection: duckdb
    lock_manager:
      class: "pydiverse.pipedag.backend.lock.ZooKeeperLockManager"
      args:
        hosts: "localhost:2181"

  snowflake:
    instance_id: pd_snowflake
    table_store:
      table_store_connection: snowflake
    lock_manager:
      class: "pydiverse.pipedag.backend.lock.ZooKeeperLockManager"
      args:
        hosts: "localhost:2181"


  ## Table Cache Instances

  local_table_cache:
    instance_id: local_table_cache
    table_store:
      local_table_cache:
        class: "pydiverse.pipedag.backend.table.cache.ParquetTableCache"
        args:
          base_path: "/tmp/pipedag/table_cache"

  local_table_cache_inout:
    instance_id: local_table_cache_inout
    table_store:
      local_table_cache:
        store_input: true
        store_output: true
        use_stored_input_as_cache: true
        class: "pydiverse.pipedag.backend.table.cache.ParquetTableCache"
        args:
          base_path: "/tmp/pipedag/table_cache"

  local_table_cache_inout_numpy:
    instance_id: local_table_cache_inout_numpy
    table_store:
      hook_args:
        pandas:
          dtype_backend: "numpy"
      local_table_cache:
        store_input: true
        store_output: true
        use_stored_input_as_cache: true
        class: "pydiverse.pipedag.backend.table.cache.ParquetTableCache"
        args:
          base_path: "/tmp/pipedag/table_cache"

  local_table_store:
    instance_id: local_table_store
    table_store:
      local_table_cache:
        store_input: true
        store_output: true
        use_stored_input_as_cache: false
        class: "pydiverse.pipedag.backend.table.cache.ParquetTableCache"
        args:
          base_path: "/tmp/pipedag/table_cache"

  ## Orchestration Engine Instances

  dask_engine:
    instance_id: dask_engine
    orchestration:
      class: "pydiverse.pipedag.engine.DaskEngine"
      args:
        num_workers: 8

  prefect_engine:
    instance_id: prefect_engine
    orchestration:
      class: "pydiverse.pipedag.engine.prefect.PrefectEngine"