Automate unstructured data ingestion with Auto-PipeLoad

Step 1: Configure credentials

-- Configure OSS access
SELECT auto_pipeload.set_oss_credentials(
    'yourAccessKeyID',      -- Replace with your AccessKey ID
    'yourAccessKeySecret',  -- Replace with your AccessKey secret
    'yourEndpoint'          -- Replace with your internal endpoint
);

-- Configure the DashScope API key (required for vectorization)
SELECT auto_pipeload.set_dashscope_api_key('sk-xxxxxxxxxxxxxxxx');

Step 2: Create a destination table

CREATE TABLE public.documents_text (
    id           SERIAL PRIMARY KEY,
    original_content TEXT,
    insert_time  TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    file_url     TEXT  -- Recommended: tracks source files for updates and deletions
);

Step 3: Register a pipeline

Call register_pipeload to define the OSS directory to monitor and the destination table. This saves the pipeline configuration — it does not trigger an immediate sync.

SELECT auto_pipeload.register_pipeload(
    p_oss_dir    => 'oss://testBucketName/testDocuments/',  -- [Required] OSS directory to monitor
    p_table_name => 'public.documents_text',                -- [Required] Destination table
    text_column  => 'original_content',                     -- [Required] Column for extracted text
    embedding    => false,                                  -- Disable vectorization
    with_chunk   => false,                                  -- Disable chunking
    intime_column => 'insert_time',                         -- Column for ingestion timestamp
    url_column   => 'file_url'                              -- Column for source file URL
);

Step 4: Upload files and trigger a refresh

1. Upload .txt, .pdf, or .docx files to oss://testBucketName/testDocuments/ using ossutil or the OSS console.

2. Trigger a manual refresh:

   SELECT auto_pipeload.refresh_pipeload(
       'oss://testBucketName/testDocuments/',
       'public.documents_text'
   );

3. Query the destination table to verify the results:

   SELECT * FROM public.documents_text LIMIT 10;

   The original_content column contains the extracted text. The file_url column records the OSS path of each source file.

4. Check pipeline status:

   SELECT * FROM auto_pipeload.show_pipeload_status();

   The status table shows pending_files, processed_files, and failed_files for each pipeline, along with last_run and last_status. If you see pending_files > 0 and processed_files = 0 immediately after triggering a refresh, the pipeline is still processing — wait a moment and query again.

Step 5: Set up auto-refresh with pg_cron

Connect to the postgres database and create a scheduled task that calls the Auto-PipeLoad scheduler once per minute. The scheduler checks all registered pipelines and triggers processing for any that are due.

pg_cron is pre-installed on all AnalyticDB for PostgreSQL instances. To view or modify scheduled tasks, see pg_cron.

-- Replace '<yourDatabaseName>' with the name of your database
SELECT cron.schedule(
    'auto-pipeload-scheduler',               -- Unique task name
    '* * * * *',                             -- Run once per minute
    'SELECT auto_pipeload.scheduled_refresh()',
    '<yourDatabaseName>'
);

How the two scheduling intervals interact: The * * * * * cron expression runs scheduled_refresh every minute, but each pipeline only processes files when its own period_minutes interval has elapsed (default: 5 minutes). Think of pg_cron as a heartbeat that wakes the scheduler, and period_minutes as the per-pipeline cooldown. For OSS directories with low update frequency, increase period_minutes when registering the pipeline to reduce unnecessary processing cycles.

Step 1: Create a vector table

The table needs columns for the text chunks and their vectors. Create an Approximate Nearest Neighbor (ANN) index on the vector column for high-performance retrieval.

CREATE TABLE public.articles (
    id          SERIAL PRIMARY KEY,
    sentence    TEXT,
    vector      REAL[],
    insert_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    file_url    TEXT
) USING heap;

-- Store vectors inline for retrieval performance
ALTER TABLE public.articles ALTER COLUMN vector SET STORAGE PLAIN;

-- ANN index for fast similarity search
-- dim must match your embedding model's output dimension
CREATE INDEX idx_articles_l2 ON articles
    USING ann(vector)
    WITH (dim=1024, distancemeasure=l2, hnsw_m=64, hnsw_ef_construction=128);

Step 2: Register a vectorization pipeline

Enable both embedding and with_chunk, and map the chunk_column and vector_column parameters to the correct table columns.

SELECT auto_pipeload.register_pipeload(
    p_oss_dir     => 'oss://testBucketName/aiDocuments/',
    p_table_name  => 'public.articles',
    embedding     => true,
    with_chunk    => true,
    chunk_column  => 'sentence',    -- Column for text chunks
    vector_column => 'vector',      -- Column for embedding vectors
    url_column    => 'file_url',
    intime_column => 'insert_time',
    chunk_size    => 1000,          -- Characters per chunk (default: 500)
    chunk_overlap => 100,           -- Overlapping characters between chunks (default: 50)
    embedding_dim => 1024           -- Must match ANN index dim
);

Choosing chunk parameters: chunk_size and chunk_overlap directly affect RAG retrieval quality:

Set chunk_overlap to roughly 10% of chunk_size to preserve continuity across chunk boundaries. The defaults (chunk_size=500, chunk_overlap=50) are a reasonable starting point for mixed content.

Step 3: Trigger a refresh

Upload files to the OSS directory, then trigger a manual refresh or wait for pg_cron to call the scheduler:

SELECT auto_pipeload.refresh_pipeload(
    'oss://testBucketName/aiDocuments/',
    'public.articles'
);

Files are extracted, chunked, sent to the embedding API, and written to the table.

Step 4: Run a similarity search

Use the <-> vector distance operator to find the chunks most similar to a query vector:

-- Find the 10 chunks most similar to the query vector
-- Replace array[...]::real[] with your 1024-dimension query vector
SELECT id, sentence, l2_distance(vector, array[...]::real[]) AS score
FROM public.articles
ORDER BY vector <-> array[...]::real[]
LIMIT 10;

set_oss_credentials

Configures and stores OSS access credentials.

auto_pipeload.set_oss_credentials(
    access_key_id     TEXT,  -- AccessKey ID
    access_key_secret TEXT,  -- AccessKey secret
    endpoint          TEXT   -- OSS internal endpoint
)

Example

SELECT auto_pipeload.set_oss_credentials(
    'access_key_id',
    'access_key_secret',
    'endpoint'
);

set_dashscope_api_key

Configures and stores the DashScope API key for vector embedding.

auto_pipeload.set_dashscope_api_key(
    api_key TEXT  -- DashScope API key
)

Example

SELECT auto_pipeload.set_dashscope_api_key('sk-xxxxxxxxxxxxxxxx');

register_pipeload

Registers a data pipeline that defines how files flow from an OSS directory to a database table.

auto_pipeload.register_pipeload(
    p_oss_dir        TEXT,     -- [Required] OSS directory path (e.g., oss://testBucketName/documents/)
    p_table_name     TEXT,     -- [Required] Destination table name
    embedding        BOOLEAN,  -- Enable vectorization. Default: false
    with_chunk       BOOLEAN,  -- Enable text chunking. Default: false
    text_column      TEXT,     -- Column for raw extracted text
    chunk_column     TEXT,     -- Column for text chunks (used when with_chunk=true)
    vector_column    TEXT,     -- Column for vector embeddings (used when embedding=true)
    url_column       TEXT,     -- Column for source file URL (strongly recommended for updates and deletions)
    intime_column    TEXT,     -- Column for data ingestion timestamp
    period_minutes   INTEGER,  -- Pipeline refresh interval in minutes. Default: 5
    chunk_size       INTEGER,  -- Characters per text chunk. Default: 500
    chunk_overlap    INTEGER,  -- Overlapping characters between chunks. Default: 50
    api_url          TEXT,     -- LLM API URL. Default: https://dashscope.aliyuncs.com/compatible-mode/v1
    embedding_model  TEXT,     -- Embedding model name. Default: text-embedding-v3
    embedding_dim    INTEGER   -- Embedding vector dimension. Default: 1024
)

Parameter reference

Example 1: Plain text extraction

SELECT auto_pipeload.register_pipeload(
    p_oss_dir    => 'oss://testBucketName/documents/',
    p_table_name => 'public.documents_text',
    text_column  => 'original_content',
    embedding    => false,
    with_chunk   => false,
    intime_column => 'insert_time',
    url_column   => 'file_url'
);

Example 2: Chunking and vectorization

SELECT auto_pipeload.register_pipeload(
    p_oss_dir     => 'oss://testBucketName/articles/',
    p_table_name  => 'public.articles',
    embedding     => true,
    with_chunk    => true,
    chunk_column  => 'sentence',
    vector_column => 'vector',
    url_column    => 'file_url',
    intime_column => 'insert_time',
    chunk_size    => 1000,
    chunk_overlap => 100,
    embedding_dim => 1024
);

refresh_pipeload

Manually triggers the sync and processing logic for a specific pipeline.

auto_pipeload.refresh_pipeload(
    p_oss_dir    TEXT,  -- OSS directory path
    p_table_name TEXT   -- Destination table name
)

Example

SELECT auto_pipeload.refresh_pipeload(
    'oss://testBucketName/testDocuments/',
    'public.documents_text'
);

reload_pipeload

Purges existing data from the destination table and performs a full resynchronization from OSS.

auto_pipeload.reload_pipeload(
    p_oss_dir        TEXT,     -- OSS directory path
    p_table_name     TEXT,     -- Destination table name
    p_truncate_table BOOLEAN   -- true = TRUNCATE (faster); false = DELETE
)

Example

-- Use TRUNCATE for a fast full reload
SELECT auto_pipeload.reload_pipeload(
    'oss://testBucketName/documents/',
    'public.documents_text',
    true
);

show_pipeload_status

Returns status information for all registered pipelines.

auto_pipeload.show_pipeload_status()

Returned fields

Example

SELECT * FROM auto_pipeload.show_pipeload_status();

pause_pipeload

Pauses a pipeline. The pipeline stops processing new files until resumed.

auto_pipeload.pause_pipeload(
    p_oss_dir    TEXT,  -- OSS directory path
    p_table_name TEXT   -- Destination table name
)

Example

SELECT auto_pipeload.pause_pipeload(
    'oss://testBucketName/documents/',
    'public.documents_text'
);

activate_pipeload

Resumes a paused pipeline.

auto_pipeload.activate_pipeload(
    p_oss_dir    TEXT,  -- OSS directory path
    p_table_name TEXT   -- Destination table name
)

Example

SELECT auto_pipeload.activate_pipeload(
    'oss://testBucketName/documents/',
    'public.documents_text'
);

unregister_pipeload

Removes a pipeline and optionally deletes the data it loaded.

auto_pipeload.unregister_pipeload(
    p_oss_dir      TEXT,     -- OSS directory path
    p_table_name   TEXT,     -- Destination table name
    p_cleanup_data BOOLEAN   -- true = delete related data; false = keep data. Default: false
)

Example

-- Remove the pipeline but keep the loaded data
SELECT auto_pipeload.unregister_pipeload(
    'oss://testBucketName/documents/',
    'public.documents_text',
    false
);

-- Remove the pipeline and delete all loaded data
SELECT auto_pipeload.unregister_pipeload(
    'oss://testBucketName/documents/',
    'public.documents_text',
    true
);

scheduled_refresh

The main scheduler function. Checks all registered pipelines and triggers processing for those whose period_minutes interval has elapsed.

auto_pipeload.scheduled_refresh()