Import Parquet files with INSERT

Applies to VAST Cluster 5.5.

Imports Parquet files that are already in a bucket on the cluster into a table. The cluster reads the files itself: their data doesn't pass through your client.

Syntax

INSERT INTO "<bucket>/<schema>"."<table> vast.import_data"
    ([<column>, ...,] "$parquet_file_path")
VALUES ([<value>, ...,] '<bucket>/<key>') [, ...]

Parameters and rules

• As with the Python SDK's import_files(), the table must have every column that's in the file, with the same type (ingest).
• The statement returns no row count. Count the rows afterwards if you need to.
• Each table remembers the files imported into it. Importing the same file again fails with File was already imported to table.
• Put the files under a key prefix that isn't the name of a schema in the same bucket; see stage the files.
• You need s3:TabularImportData on the database, as for the SDK's import (ingest). Admin task: send to your platform team.
• The Query Engine can't import into a partitioned table (not supported).

The form with the plain table name, INSERT INTO <table> (..., "$parquet_file_path") VALUES (...), fails with Mismatched column name between file to import and existing table. Use the vast.import_data form above.

What happens on the cluster

The Query Engine passes the paths to VAST DataBase, and the cluster reads each Parquet file from the bucket and writes its rows into the table. Your client sends only the statement. It works like the Python SDK's import_files() (write path).

Example

The examples use the run() helper and the environment variables described in the SQL reference. To put the files in the bucket, this page also needs boto3 (pip install boto3 pyarrow) and the cluster's S3 endpoint in AWS_S3_ENDPOINT_URL, as in the ingest guide. This block connects, creates the schema docs_sql_import_parquet with a table, and stages two Parquet files:

import io

import boto3
import pyarrow as pa
import pyarrow.parquet as pq

import os
import re

import adbc_driver_manager
import adbc_driver_manager.dbapi as dbapi
import adbc_driver_vastdb

BUCKET = os.environ["VASTDB_BUCKET"]
SCHEMA = f'"{BUCKET}/docs_sql_import_parquet"'

def connect(autocommit=True):
    return dbapi.connect(
        driver=adbc_driver_vastdb.get_driver_path(),
        db_kwargs={
            "vast.db.endpoint": os.environ["QE_ENDPOINT"],
            "vast.db.access_key": os.environ["AWS_ACCESS_KEY_ID"],
            "vast.db.secret_key": os.environ["AWS_SECRET_ACCESS_KEY"],
        },
        autocommit=autocommit,
    )

def server_message(e):
    """The server's message in a driver error, without the request ID or the echoed statement."""
    text = re.sub(r"\[req=0x[0-9a-f]+\]: ", "", str(e))
    found = (re.search(r'message: "(.*?)", request_id', text, re.S)
             or re.search(r"description: (.*)\)", text, re.S))
    message = found.group(1) if found else text.strip().splitlines()[-1]
    message = re.sub(r"^Plan query: '.*?' failed with: ", "", message, flags=re.S)
    message = re.split(r"\\n|\n", message)[0].replace('\\"', '"')
    return message.replace(BUCKET, "<bucket>")

def run(sql):
    """Run one statement. Print the rows it returns, or the error."""
    try:
        with conn.cursor() as cur:
            cur.execute(sql)
            rows = cur.fetchall() if cur.description else []
    except (adbc_driver_manager.Error, OSError) as e:
        print(f"{type(e).__name__}: {server_message(e)}")
    else:
        print(rows)

conn = connect()
TABLE = f"{SCHEMA}.readings"
run(f"CREATE SCHEMA IF NOT EXISTS {SCHEMA}")
run(f"CREATE TABLE IF NOT EXISTS {TABLE} (id BIGINT, station VARCHAR)")

s3 = boto3.client("s3", endpoint_url=os.environ["AWS_S3_ENDPOINT_URL"])
PREFIX = "staging/docs_sql_import_parquet/"

def stage(name, table):
    buf = io.BytesIO()
    pq.write_table(table, buf)
    s3.put_object(Bucket=BUCKET, Key=PREFIX + name, Body=buf.getvalue())
    return f"{BUCKET}/{PREFIX}{name}"

full = stage("full.parquet", pa.table({"id": pa.array([1, 2], pa.int64()), "station": ["north", "south"]}))
ids_only = stage("ids-only.parquet", pa.table({"id": pa.array([3, 4], pa.int64())}))
print(full.replace(BUCKET, "<bucket>"), ids_only.replace(BUCKET, "<bucket>"))

[]
[]
<bucket>/staging/docs_sql_import_parquet/full.parquet <bucket>/staging/docs_sql_import_parquet/ids-only.parquet

Import the first file. Then import the second, which has no station column, and give station a value for its rows:

IMPORT = f'"{BUCKET}/docs_sql_import_parquet"."readings vast.import_data"'
run(f"""INSERT INTO {IMPORT} ("$parquet_file_path") VALUES ('{full}')""")
run(f"""INSERT INTO {IMPORT} (station, "$parquet_file_path") VALUES ('east', '{ids_only}')""")
run(f"SELECT id, station FROM {TABLE} ORDER BY id LIMIT 10")

[]
[]
[(1, 'north'), (2, 'south'), (3, 'east'), (4, 'east')]

Importing the same file again fails. So do the form with the plain table name, and a path with a leading /:

run(f"""INSERT INTO {IMPORT} ("$parquet_file_path") VALUES ('{full}')""")
run(f"""INSERT INTO {TABLE} ("$parquet_file_path") VALUES ('{full}')""")
run(f"""INSERT INTO {IMPORT} ("$parquet_file_path") VALUES ('/{full}')""")

InternalError: internal error: File was already imported to table
InternalError: internal error: Mismatched column name between file to import and existing table.
InternalError: internal error: The specified key does not exist.

Clean up: drop the table and the schema, and delete the staged files:

run(f"DROP TABLE IF EXISTS {TABLE}")
run(f"DROP SCHEMA IF EXISTS {SCHEMA}")
conn.close()
for name in ("full.parquet", "ids-only.parquet"):
    s3.delete_object(Bucket=BUCKET, Key=PREFIX + name)
print("Deleted the staged files")

[]
[]
Deleted the staged files

Errors

The class is InternalError.

Trino and Spark

Trino and Spark import with the same suffix, written with parentheses, and take paths that start with a slash; see ingest.

See also: ingest, INSERT.