### Run Basic SQLAlchemy Examples

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/README.md

Executes basic examples demonstrating how to use SQLAlchemy with SQL literal text against a Kinetica database. Requires Kinetica URL, username, password, schema, and an SSL certificate bypass flag.

```bash
cd examples
python3 basic_examples.py <kinetica_url> <username> <password> <schema> <bypass_ssl_cert_check>
```

--------------------------------

### Run Kinetica Dialect for SQLAlchemy Examples

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/README.md

Executes examples showcasing advanced SQL and Kinetica-specific features using the Kinetica Dialect for SQLAlchemy. Requires Kinetica URL, username, password, schema, an SSL certificate bypass flag, and a schema recreation flag.

```bash
cd examples
python3 sqlalchemy_api_examples.py <kinetica_url> <username> <password> <schema> <bypass_ssl_cert_check> <recreate_schema>
```

--------------------------------

### ASOF Join for Time-Series Data with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

This example shows how to implement an ASOF (As Of) join for time-series data using SQLAlchemy and the `sqlalchemy_kinetica.custom_commands.Asof` class. It matches trades with the closest preceding quotes within a specified time interval. Ensure `sqlalchemy_kinetica` is installed.

```python
from sqlalchemy_kinetica.custom_commands import Asof

metadata = MetaData()

with engine.connect() as conn:
    quotes = Table('quotes', metadata, autoload_with=conn, schema="my_schema").alias("q")
    trades = Table('trades', metadata, autoload_with=conn, schema="my_schema").alias("t")
    
    # Define the ASOF condition
    asof_condition = Asof(
        trades.c.dt,           # Left column (trade datetime)
        quotes.c.open_dt,      # Right column (quote datetime)
        text("INTERVAL '-1' DAY"),  # Relative range begin
        text("INTERVAL '0' DAY"),   # Relative range end
        text("MAX")            # MIN or MAX for tie-breaking
    )
    
    # Build the query
    query = select(
        trades.c.id,
        trades.c.dt.label('execution_dt'),
        quotes.c.open_dt.label('quote_dt'),
        trades.c.price.label('execution_price'),
        quotes.c.open_price
    ).select_from(
        trades.outerjoin(
            quotes, 
            (trades.c.ticker == quotes.c.symbol) & asof_condition
        )
    )
    
    # Compile with literal_binds
    compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
    result = conn.execute(compiled)
```

--------------------------------

### Perform Basic UPDATE Operation with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates a simple UPDATE statement in SQLAlchemy. This example shows how to update specific columns for rows that match a given condition.

```python
from sqlalchemy import update, MetaData, Table

metadata = MetaData()

with engine.connect() as conn:
    employees = Table('employees', metadata, autoload_with=conn, schema="my_schema")
    
    # Simple update
    stmt = update(employees).where(
        employees.c.id == 5
    ).values(
        salary=75000,
        department="Sales"
    )
    
    conn.execute(stmt)
```

--------------------------------

### Work with SQLAlchemy Connections

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to establish and use connections to the Kinetica database via SQLAlchemy. It shows usage with context managers for automatic closing and transaction blocks using `begin()`.

```python
# Using context manager (recommended)
with engine.connect() as conn:
    result = conn.execute(select(my_table))
    for row in result:
        print(row)

# Using begin() for transaction blocks
with engine.begin() as conn:
    conn.execute(insert(my_table).values(id=1, name="test"))
```

--------------------------------

### Create SQLAlchemy Engine using Environment Variables

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Creates a SQLAlchemy engine for Kinetica by reading connection details from environment variables. This promotes secure and flexible configuration.

```python
import os

ENV_URL = os.getenv("KINETICA_URL", "http://localhost:9191")
ENV_USER = os.getenv("KINETICA_USER", "")
ENV_PASS = os.getenv("KINETICA_PASS", "")
ENV_SCHEMA = os.getenv("KINETICA_SCHEMA", "default")

engine = create_engine(
    "kinetica://",
    connect_args={
        "url": ENV_URL,
        "username": ENV_USER,
        "password": ENV_PASS,
        "default_schema": ENV_SCHEMA,
    }
)
```

--------------------------------

### Configure Kinetica Indexes in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Provides examples for defining various index types in Kinetica using SQLAlchemy. This includes standard multi-column indexes, chunk skip indexes, geospatial indexes, and CAGRA indexes for vector similarity search. The `kinetica_index`, `kinetica_chunk_skip_index`, `kinetica_geospatial_index`, and `kinetica_cagra_index` arguments are used.

```python
indexed_table = Table(
    "indexed_data",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("dept_id", Integer),
    Column("name", VARCHAR(100)),
    Column("location", GEOMETRY),
    Column("coords_lon", REAL),
    Column("coords_lat", REAL),
    Column("embedding", VECTOR(128)),
    schema="my_schema",
    
    # Standard index (list of lists for multiple indexes)
    kinetica_index=[["dept_id"], ["name"]],
    
    # Chunk skip index (single column name)
    kinetica_chunk_skip_index="id",
    
    # Geospatial indexes
    kinetica_geospatial_index=[
        ["location"],
        ["coords_lon", "coords_lat"]
    ],
    
    # CAGRA index for vector similarity search
    kinetica_cagra_index=["embedding"]
)
```

--------------------------------

### SQL RANK and PERCENT_RANK Window Functions in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Shows how to implement RANK and PERCENT_RANK window functions using SQLAlchemy. These functions assign a rank to each row within a partition based on a specified order.

```python
from sqlalchemy import MetaData, select, func, Table, cast, FLOAT

metadata = MetaData()

with engine.connect() as conn:
    nyctaxi = Table('nyctaxi', metadata, autoload_with=conn, schema='demo')
    
    # Rank function
    ranked_fare = func.rank().over(
        partition_by=nyctaxi.c.vendor_id,
        order_by=nyctaxi.c.total_amount
    ).label('ranked_fare')
    
    # Percent rank (multiply by 100)
    percent_ranked = func.percent_rank().over(
        partition_by=nyctaxi.c.vendor_id,
        order_by=nyctaxi.c.total_amount
    ) * 100
    
    query = select(
        nyctaxi.c.vendor_id,
        nyctaxi.c.total_amount.label('fare'),
        ranked_fare,
        cast(percent_ranked, FLOAT).label('percent_ranked_fare')
    ).where(
        nyctaxi.c.passenger_count == 3
    )
    
    # Must compile with literal_binds for arithmetic with literals
    compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
    result = conn.execute(compiled)
```

--------------------------------

### Select with JOIN Operations (Python)

Source: https://context7.com/kineticadb/sqlalchemy-kinetica/llms.txt

Illustrates how to perform inner and outer joins between tables using SQLAlchemy's expressive join syntax. This example includes aliasing for self-joins and ordering results.

```python
from sqlalchemy import MetaData, Table, select, nullsfirst, asc

metadata = MetaData()

with engine.connect() as conn:
    employee = Table('employee', metadata, autoload_with=conn, schema="my_schema")

    # Aliases for self-join
    e = employee.alias("e")
    m = employee.alias("m")

    # Left outer join with ordering
    query = select(
        (e.c.last_name + ', ' + e.c.first_name).label("Employee_Name"),
        (m.c.last_name + ', ' + m.c.first_name).label("Manager_Name")
    ).select_from(
        e.outerjoin(m, e.c.manager_id == m.c.id)
    ).where(
        e.c.dept_id.in_([1, 2, 3])
    ).order_by(
        nullsfirst(asc(m.c.id)),
        e.c.hire_date
    )

    result = conn.execute(query)
    for row in result:
        print(f"{row.Employee_Name} reports to {row.Manager_Name}")
```

--------------------------------

### Configuring Kinetica Engine with SSL and Timeout

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to configure the SQLAlchemy engine for Kinetica connections, including settings for bypassing SSL certificate checks for self-signed certificates and adjusting connection timeouts.

```python
from sqlalchemy import create_engine

# Example with SSL bypass and increased timeout
engine = create_engine("kinetica://", connect_args={
    "url": "https://your-server:9191",
    "bypass_ssl_cert_check": True,  # For self-signed certs
    "timeout": 120,  # Increase timeout in seconds
    ...
})
```

--------------------------------

### Install Kinetica Dialect for SQLAlchemy using pip

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/README.md

Installs the Kinetica Dialect for SQLAlchemy package using pip. This is the standard method for adding the dialect to your Python environment. Ensure you have pip installed and updated.

```bash
pip3 install sqlalchemy-kinetica
```

--------------------------------

### Autoloading Existing Tables with SQLAlchemy and Kinetica

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates how to use `autoload_with` to accurately reflect the schema of existing Kinetica tables within SQLAlchemy, preventing potential mismatches and ensuring correct data handling.

```python
from sqlalchemy import Table, MetaData

# Good - gets accurate schema from database
metadata = MetaData()
employees = Table('employees', metadata, autoload_with=conn, schema='my_schema')

# Avoids schema mismatches
```

--------------------------------

### Optimized Batch Inserts with `ki_insert`

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Compares the efficient batch insertion method using `ki_insert` with the less efficient row-by-row insertion for SQLAlchemy and Kinetica. `ki_insert` is recommended for performance when inserting large datasets.

```python
# Good - optimized for batch inserts
conn.execute(ki_insert(employees), list_of_records)

# Less efficient for large batches
for record in list_of_records:
    conn.execute(insert(employees).values(**record))
```

--------------------------------

### Import SQLAlchemy and Kinetica Components

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Imports necessary components from SQLAlchemy and the Kinetica dialect. This includes core SQLAlchemy functions, DDL tools, Kinetica-specific types, and custom commands.

```python
from sqlalchemy import (
    create_engine, MetaData, Table, Column, Integer, String,
    select, insert, update, delete, func, and_, or_, text,
    cast, case, literal, column, alias, union, intersect, except_
)
from sqlalchemy.sql.ddl import CreateTable

# Kinetica-specific imports
from sqlalchemy_kinetica.dialect import KineticaDialect
from sqlalchemy_kinetica.kinetica_types import (
    TINYINT, SMALLINT, BIGINT, UnsignedBigInteger,
    FLOAT, DOUBLE, REAL, DECIMAL,
    VARCHAR, BLOB, JSON, JSONArray,
    DATE, TIME, DATETIME, TIMESTAMP,
    IPV4, UUID, VECTOR, GEOMETRY, BlobWKT
)
from sqlalchemy_kinetica.custom_commands import (
    ki_insert, KiUpdate, CreateTableAs, Asof, FirstValue,
    PivotSelect, UnpivotSelect, FilterByString, EvaluateModel
)
```

--------------------------------

### SQL Aggregations Without GROUP BY in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Shows how to perform aggregate functions like AVG, COUNT, MIN, and MAX on an entire dataset without using a GROUP BY clause. This is useful for getting overall statistics.

```python
from sqlalchemy import select, func, Table

# Assuming 'employees' table is already defined and connected via 'conn'
# employees = Table('employees', metadata, autoload_with=conn)

query = select(
    func.round(func.avg(employees.c.salary), 2).label('avg_salary'),
    func.count().label('total_count'),
    func.min(employees.c.salary).label('min_salary'),
    func.max(employees.c.salary).label('max_salary')
)

# Compile with literal_binds for numeric arguments
compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
result = conn.execute(compiled)
```

--------------------------------

### Perform UPDATE with Subquery using SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates how to use a subquery within an UPDATE statement in SQLAlchemy. This example updates employee salaries based on a correlated subquery that calculates the maximum salary in each department.

```python
from sqlalchemy import alias, func, select, update, literal

metadata = MetaData()

with engine.connect() as conn:
    e_lookup = Table('employee', metadata, autoload_with=conn, schema="my_schema")
    e_base = alias(e_lookup, name="b")
    
    # Correlated subquery
    max_sal_in_dept = (
        select(func.max(e_lookup.c.sal))
        .where(e_base.c.dept_id == e_lookup.c.dept_id)
    ).scalar_subquery()
    
    # Update with subquery in SET
    stmt = update(e_base).values(
        sal=max_sal_in_dept * literal(0.1) + e_base.c.sal * literal(0.9)
    )
    
    compiled = stmt.compile(conn, compile_kwargs={"literal_binds": True})
    conn.execute(compiled)
```

--------------------------------

### SQL FIRST_VALUE with IGNORE NULLS in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates using the FIRST_VALUE window function with the IGNORE NULLS option in SQLAlchemy. This calculates the difference from the lowest and highest non-null tip amounts within each vendor partition.

```python
from sqlalchemy_kinetica.custom_commands import FirstValue
from sqlalchemy import select, func, Table, desc

# Assuming 'nyctaxi' table is already defined and connected via 'conn'
# nyctaxi = Table('nyctaxi', metadata, autoload_with=conn, schema='demo')

tip_amount = nyctaxi.c.tip_amount

# FIRST_VALUE with IGNORE NULLS
lowest_tip = tip_amount - FirstValue(tip_amount, ignore_nulls=True).over(
    partition_by=nyctaxi.c.vendor_id,
    order_by=tip_amount
)

highest_tip = tip_amount - FirstValue(tip_amount, ignore_nulls=True).over(
    partition_by=nyctaxi.c.vendor_id,
    order_by=desc(tip_amount)
)

query = select(
    nyctaxi.c.vendor_id,
    tip_amount,
    lowest_tip.label('diff_from_lowest'),
    highest_tip.label('diff_from_highest')
)

compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
result = conn.execute(compiled)
```

--------------------------------

### Inner Join Operations with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

This example illustrates how to perform an inner join using SQLAlchemy, specifically a self-join with aliases. It selects employee and manager names by joining the 'employee' table to itself on the manager ID. The `select_from` method is used to specify the join condition.

```python
metadata = MetaData()

with engine.connect() as conn:
    employee = Table('employee', metadata, autoload_with=conn, schema="my_schema")
    
    # Self-join with aliases
    e = employee.alias("e")
    m = employee.alias("m")
    
    query = select(
        (e.c.last_name + ', ' + e.c.first_name).label("Employee_Name"),
        (m.c.last_name + ', ' + m.c.first_name).label("Manager_Name")
    ).select_from(
        e.join(m, e.c.manager_id == m.c.id)
    )
    
    result = conn.execute(query)
```

--------------------------------

### Create Basic SQLAlchemy Table

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates the fundamental process of creating a SQLAlchemy Table with basic column types like Integer, VARCHAR, and DECIMAL. It also shows how to initiate table creation using `metadata.create_all(engine)`. Assumes an `engine` object is already configured.

```python
from sqlalchemy import MetaData, Table, Column, Integer, String

metadata = MetaData()

employees = Table(
    "employees",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", VARCHAR(100)),
    Column("department", VARCHAR(50)),
    Column("salary", DECIMAL(18, 4)),
    schema="my_schema"
)

# Create the table
metadata.create_all(engine)
```

--------------------------------

### Basic Select Queries with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

This snippet demonstrates fundamental select operations in SQLAlchemy, including selecting all columns, specific columns, filtering with WHERE clauses, ordering results, and applying LIMIT and OFFSET. It assumes the 'employees' table is available.

```python
from sqlalchemy import select, func

metadata = MetaData()

with engine.connect() as conn:
    employees = Table('employees', metadata, autoload_with=conn, schema="my_schema")
    
    # Simple select all
    query = select(employees)
    result = conn.execute(query)
    
    # Select specific columns
    query = select(employees.c.name, employees.c.salary)
    
    # With WHERE clause
    query = select(employees).where(employees.c.salary > 50000)
    
    # With ORDER BY
    query = select(employees).order_by(employees.c.salary.desc())
    
    # With LIMIT
    query = select(employees).limit(10)
    
    # With OFFSET
    query = select(employees).limit(10).offset(20)
```

--------------------------------

### SQL Aggregations With GROUP BY in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates performing aggregate functions grouped by a specific column (e.g., department) and applying a HAVING clause to filter these groups. Also includes ordering the results.

```python
from sqlalchemy import select, func, Table

# Assuming 'employees' table is already defined and connected via 'conn'
# employees = Table('employees', metadata, autoload_with=conn)

query = select(
    employees.c.department,
    func.count().label('emp_count'),
    func.avg(employees.c.salary).label('avg_salary')
).group_by(
    employees.c.department
).having(
    func.count() > 5
).order_by(
    employees.c.department
)

compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
result = conn.execute(compiled)
```

--------------------------------

### Create PIVOT Query with SQLAlchemy Kinetica

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to construct a PIVOT query using the PivotSelect class from sqlalchemy_kinetica.custom_commands. This involves specifying columns for pivoting, the aggregate function, the pivot column, and the pivot values.

```python
from sqlalchemy_kinetica.custom_commands import PivotSelect
from sqlalchemy import column, MetaData, Table

metadata = MetaData()

with engine.connect() as conn:
    phone_list = Table('phone_list', metadata, autoload_with=conn, schema="my_schema")
    
    # Create PIVOT query
    query = (
        PivotSelect(
            column("name"),
            column("Home_Phone"),
            column("Work_Phone"),
            column("Cell_Phone")
        )
        .select_from(phone_list)
        .pivot(
            "max(phone_number) AS Phone",  # Aggregate function
            "phone_type",                   # Pivot column
            ["'Home'", "'Work'", "'Cell'"]  # Pivot values
        )
        .order_by(column("name"))
    )
    
    compiled = query.compile(conn)
    result = conn.execute(compiled)
```

--------------------------------

### Create Optimized Kinetica Tables with Sharding and Partitioning

Source: https://context7.com/kineticadb/sqlalchemy-kinetica/llms.txt

Defines a Kinetica table with advanced configurations including sharding, partitioning, tier strategies, and various index types. This example showcases how to set table properties, define a range-based partition clause, and specify shard keys, geospatial indexes, and CAGRA indexes for vector search.

```python
from sqlalchemy import MetaData, Table, Column, Integer
from sqlalchemy_kinetica.kinetica_types import VARCHAR, DECIMAL, DATE, BlobWKT, REAL, VECTOR

metadata = MetaData()

table_properties = {
    "CHUNK SIZE": 1000000,
    "NO_ERROR_IF_EXISTS": "TRUE",
    "TTL": 120,
}

partition_clause = """
PARTITION BY RANGE (YEAR(hire_date))
PARTITIONS
(
    order_2020 MIN(2020) MAX(2021),
    order_2021               MAX(2022),
    order_2022               MAX(2023),
    order_2023               MAX(2024)
)
"""

employee = Table(
    "employee",
    metadata,
    Column("id", Integer, nullable=False, primary_key=True),
    Column("dept_id", Integer, nullable=False, primary_key=True,
           info={"shard_key": True}),  # Mark as shard key
    Column("manager_id", Integer),
    Column("first_name", VARCHAR(30)),
    Column("last_name", VARCHAR(30)),
    Column("sal", DECIMAL(10, 2)),
    Column("hire_date", DATE),
    Column("work_district", BlobWKT),
    Column("office_longitude", REAL),
    Column("office_latitude", REAL),
    Column("profile", VECTOR(10)),
    schema="my_schema",
    prefixes=["OR REPLACE"],
    info=table_properties,
    kinetica_index=[["dept_id"]],
    kinetica_chunk_skip_index="id",
    kinetica_geospatial_index=[["work_district"], ["office_longitude", "office_latitude"]],
    kinetica_cagra_index=["profile"],
    kinetica_tier_strategy="( ( VRAM 1, RAM 7, PERSIST 5 ) )",
    kinetica_partition_clause=partition_clause
)

metadata.create_all(engine)
```

--------------------------------

### Apply Tier Strategy to SQLAlchemy Table

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to define a tier strategy for a Kinetica table using the `kinetica_tier_strategy` argument in SQLAlchemy. This allows specifying storage tiers (VRAM, RAM, PERSIST) and their durations for data.

```python
tiered_table = Table(
    "tiered_data",
    metadata,
    Column("id", Integer, primary_key=True),
    schema="my_schema",
    kinetica_tier_strategy="( ( VRAM 1, RAM 7, PERSIST 5 ) )"
)
```

--------------------------------

### Troubleshooting Arithmetic Type Errors with `literal_binds`

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Provides solutions for the SQLAlchemy error `Cannot apply '*' to arguments of type '<DOUBLE> * <UNKNOWN>'` by demonstrating the use of `literal_binds=True` during query compilation or by using `sqlalchemy.literal()` for inline values.

```python
from sqlalchemy import literal, select

# Assuming 'query' is a SQLAlchemy select statement and 'conn' is the connection

# Solution 1: Compile with literal_binds=True
compiled = query.compile(conn, compile_kwargs={'literal_binds': True})
result = conn.execute(compiled)

# Solution 2: Use literal() for inline values
query = select(column * literal(100))
```

--------------------------------

### Create SQLAlchemy Tables with Prefixes

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Shows how to define SQLAlchemy Tables with Kinetica-specific prefixes like 'REPLICATED', 'OR REPLACE', and 'TEMP'. These prefixes modify table behavior, such as replication or temporary storage. The `prefixes` argument is passed directly to the `Table` constructor.

```python
# Replicated table
replicated_table = Table(
    "lookup_data",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("value", VARCHAR(100)),
    schema="my_schema",
    prefixes=["REPLICATED"]
)

# OR REPLACE table
replaceable_table = Table(
    "my_table",
    metadata,
    Column("id", Integer, primary_key=True),
    schema="my_schema",
    prefixes=["OR REPLACE"]
)

# Temporary table
temp_table = Table(
    "temp_data",
    metadata,
    Column("id", Integer, primary_key=True),
    schema="my_schema",
    prefixes=["TEMP"]
)
```

--------------------------------

### Create SQLAlchemy Engine for Kinetica

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Creates a SQLAlchemy engine instance to connect to a Kinetica database. It uses the 'kinetica://' URL scheme and specifies connection arguments like URL, username, password, and default schema.

```python
from sqlalchemy import create_engine

engine = create_engine(
    "kinetica://",
    connect_args={
        "url": "http://localhost:9191",
        "username": "admin",
        "password": "your_password",
        "default_schema": "my_schema",
        "bypass_ssl_cert_check": True,  # Optional: for self-signed certs
    }
)
```

--------------------------------

### Create Kinetica Table As (CTAS)

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to create a new Kinetica table based on the results of a SELECT query using the `CreateTableAs` command. This is useful for creating materialized views or backups. Supports options like OR REPLACE, REPLICATED, and TEMP.

```python
from sqlalchemy import MetaData, Table, select
from sqlalchemy.engine import create_engine
from sqlalchemy_kinetica.custom_commands import CreateTableAs

# Assume engine is defined elsewhere
# engine = create_engine('kinetica://user:password@host:port/database')

metadata = MetaData()

# with engine.connect() as conn:
#     source_table = Table('employee', metadata, autoload_with=conn, schema="my_schema")
    
#     # Create a new table from a SELECT query
#     create_stmt = CreateTableAs(
#         '"my_schema"."employee_backup"',
#         select(source_table),
#         prefixes=["OR REPLACE", "REPLICATED", "TEMP"]
#     )
    
#     compiled_stmt = create_stmt.compile(conn)
#     conn.execute(compiled_stmt)

```

--------------------------------

### SQL NTILE Window Function in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates the use of the NTILE window function in SQLAlchemy to divide rows into a specified number of groups (e.g., quartiles). It then calculates the average amount for each vendor, including the interquartile range.

```python
from sqlalchemy import select, func, Table, case

# Assuming 'nyctaxi' table is already defined and connected via 'conn'
# nyctaxi = Table('nyctaxi', metadata, autoload_with=conn, schema='demo')

# Subquery with NTILE
quartile_subquery = select(
    nyctaxi.c.vendor_id,
    nyctaxi.c.total_amount,
    func.ntile(4).over(
        partition_by=nyctaxi.c.vendor_id,
        order_by=nyctaxi.c.total_amount
    ).label('quartile')
).subquery()

# Main query using interquartile range
query = select(
    quartile_subquery.c.vendor_id,
    func.decimal(func.avg(quartile_subquery.c.total_amount)).label('avg_amount'),
    func.decimal(
        func.avg(
            case(
                (quartile_subquery.c.quartile.in_([2, 3]), 
                 quartile_subquery.c.total_amount),
                else_=None
            )
        )
    ).label('avg_interquartile_amount')
).group_by(
    quartile_subquery.c.vendor_id
)

compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
result = conn.execute(compiled)
```

--------------------------------

### Correctly Compiling Updates Without `literal_binds`

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Addresses the `No literal value renderer is available for literal value` error by showing the correct way to compile UPDATE statements in SQLAlchemy for Kinetica, emphasizing the exclusion of `literal_binds=True` for basic value assignments.

```python
# Wrong - causes error with literal_binds=True
stmt = employees.update().values(salary=70000)
conn.execute(stmt.compile(conn, compile_kwargs={'literal_binds': True}))

# Right - correct compilation for basic updates
stmt = employees.update().values(salary=70000)
conn.execute(stmt.compile(conn))
```

--------------------------------

### Set Kinetica Table Properties via Info

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to configure Kinetica table properties, such as 'CHUNK SIZE', 'NO_ERROR_IF_EXISTS', and 'TTL', using the `info` parameter in the SQLAlchemy `Table` definition. These properties are passed as a dictionary.

```python
table_properties = {
    "CHUNK SIZE": 1000000,
    "NO_ERROR_IF_EXISTS": "TRUE",
    "TTL": 120,
}

my_table = Table(
    "my_table",
    metadata,
    Column("id", Integer, primary_key=True),
    schema="my_schema",
    info=table_properties  # Pass properties via info parameter
)
```

--------------------------------

### Define and Use CTEs in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to define a Common Table Expression (CTE) using SQLAlchemy's select construct and then use it in a main query. This is useful for breaking down complex queries into more manageable, readable parts.

```python
from sqlalchemy import MetaData, select, func, Table

metadata = MetaData()

with engine.connect() as conn:
    employee = Table('employee', metadata, autoload_with=conn, schema="my_schema")
    
    # Define the CTE
    dept_salary_cte = (
        select(
            employee.c.manager_id,
            employee.c.sal.label('salary')
        )
        .where(employee.c.dept_id == 2)
        .cte(name='dept2_emp_sal_by_mgr')
    )
    
    # Use the CTE in main query
    query = select(
        dept_salary_cte.c.manager_id.label('mgr_id'),
        func.max(dept_salary_cte.c.salary).label('max_salary'),
        func.count().label('emp_count')
    ).group_by(
        dept_salary_cte.c.manager_id
    )
    
    compiled = query.compile(conn)
    result = conn.execute(compiled)
```

--------------------------------

### Basic Kinetica Insert Operation

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Performs a single-record insert into a Kinetica table using SQLAlchemy's `insert` construct. This is suitable for individual data entries. Assumes the table and engine are already defined.

```python
from sqlalchemy import insert, MetaData, Table
from sqlalchemy.engine import create_engine

# Assume metadata and engine are defined elsewhere
# metadata = MetaData()
# engine = create_engine('kinetica://user:password@host:port/database')

# with engine.connect() as conn:
#     employees = Table('employees', metadata, autoload_with=conn, schema="my_schema")
    
#     # Single record insert
#     stmt = insert(employees).values(
#         id=1,
#         name="John Doe",
#         department="Engineering",
#         salary=75000.00
#     )
#     conn.execute(stmt)

```

--------------------------------

### Define Kinetica Table with All Options

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Defines a Kinetica table with comprehensive options including primary keys, shard keys, schema, prefixes, chunk size, error handling, TTL, and various Kinetica-specific indexes and tier strategies. It then prints the generated DDL and creates the table.

```python
from sqlalchemy import Table, Column, Integer, VARCHAR, DECIMAL, DATE, REAL, MetaData, BlobWKT, VECTOR
from sqlalchemy.schema import CreateTable
from sqlalchemy.engine import create_engine

# Assume metadata and engine are defined elsewhere
# metadata = MetaData()
# engine = create_engine('kinetica://user:password@host:port/database')

# Placeholder for partition_clause if needed
partition_clause = "PARTITION BY RANGE (hire_date)"

employee = Table(
    "employee",
    metadata,
    Column("id", Integer, nullable=False, primary_key=True),
    Column("dept_id", Integer, nullable=False, primary_key=True, 
           info={"shard_key": True}),
    Column("manager_id", Integer),
    Column("first_name", VARCHAR(30)),
    Column("last_name", VARCHAR(30)),
    Column("sal", DECIMAL(18, 4)),
    Column("hire_date", DATE),
    Column("work_district", BlobWKT),
    Column("office_longitude", REAL),
    Column("office_latitude", REAL),
    Column("profile", VECTOR(10)),
    schema="my_schema",
    prefixes=["OR REPLACE"],
    info={
        "CHUNK SIZE": 1000000,
        "NO_ERROR_IF_EXISTS": "TRUE",
        "TTL": 120
    },
    kinetica_index=[["dept_id"]],
    kinetica_chunk_skip_index="id",
    kinetica_geospatial_index=[
        ["work_district"], 
        ["office_longitude", "office_latitude"]
    ],
    kinetica_cagra_index=["profile"],
    kinetica_tier_strategy="( ( VRAM 1, RAM 7, PERSIST 5 ) )",
    kinetica_partition_clause=partition_clause
)

# Print the DDL
# print(CreateTable(employee).compile(engine))

# Create the table
# metadata.create_all(engine)

```

--------------------------------

### Perform UNION ALL Operation with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to combine the results of two SELECT statements using the `union_all` function from SQLAlchemy. Unlike `union`, this operation includes all rows from both result sets, including duplicates.

```python
from sqlalchemy import union_all, select

# Assuming 'lunch' and 'dinner' tables and 'conn' are defined as in the UNION example
# lunch = Table('lunch_menu', metadata, autoload_with=conn, schema="my_schema")
# dinner = Table('dinner_menu', metadata, autoload_with=conn, schema="my_schema")

union_all_query = union_all(
    select(lunch.c.item, lunch.c.price),
    select(dinner.c.item, dinner.c.price)
)

compiled = union_all_query.compile(conn)
result = conn.execute(compiled)
```

--------------------------------

### Kinetica Insert with Upsert Hint

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Performs an insert operation with an upsert hint (`KI_HINT_UPDATE_ON_EXISTING_PK`) using `ki_insert`. If a record with a matching primary key already exists, it will be updated; otherwise, it will be inserted. This is useful for synchronizing data.

```python
from sqlalchemy import MetaData, Table
from sqlalchemy.engine import create_engine
from sqlalchemy_kinetica.custom_commands import ki_insert

# Assume metadata, engine, and employees table object are defined elsewhere
# metadata = MetaData()
# engine = create_engine('kinetica://user:password@host:port/database')
# employees = Table('employees', metadata, autoload_with=engine.connect(), schema="my_schema")

# Upsert mode - update existing records with matching primary key
# insert_stmt = ki_insert(employees, insert_hint="KI_HINT_UPDATE_ON_EXISTING_PK")

# records = [
#     {"id": 1, "name": "Anne Updated", "department": "Sales", "salary": 110000},
# ]

# conn.execute(insert_stmt, records)

```

--------------------------------

### Basic DELETE Operations with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Shows how to perform basic DELETE operations on a table using SQLAlchemy. This includes deleting specific records based on a single condition and deleting records that meet multiple criteria using the `and_` operator.

```python
from sqlalchemy import delete, MetaData, Table, and_

metadata = MetaData()

with engine.connect() as conn:
    employees = Table('employees', metadata, autoload_with=conn, schema="my_schema")
    
    # Delete specific records
    stmt = delete(employees).where(employees.c.id == 5)
    conn.execute(stmt)
    
    # Delete with multiple conditions
    stmt = delete(employees).where(
        and_(
            employees.c.department == "Sales",
            employees.c.salary < 30000
        )
    )
    conn.execute(stmt)
```

--------------------------------

### Define Partitioning Clause for SQLAlchemy Table

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Shows how to specify a Kinetica partitioning clause for a SQLAlchemy Table. The `kinetica_partition_clause` argument accepts a raw SQL string defining the partitioning strategy, such as range-based partitioning by year.

```python
partition_clause = """
PARTITION BY RANGE (YEAR(hire_date))
PARTITIONS
(
    orders_2020 MIN(2020) MAX(2021),
    orders_2021               MAX(2022),
    orders_2022               MAX(2023),
    orders_2023               MAX(2024)
)
"""

partitioned_table = Table(
    "orders",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("hire_date", DATE),
    schema="my_schema",
    kinetica_partition_clause=partition_clause
)
```

--------------------------------

### DELETE with Subquery using SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to perform a DELETE operation using a correlated subquery with SQLAlchemy. This is useful for deleting records based on complex conditions that require evaluating another query, such as finding the maximum ID within each department.

```python
from sqlalchemy import delete, func, MetaData, Table, select

metadata = MetaData()

with engine.connect() as conn:
    e_base = Table('employee', metadata, autoload_with=conn, schema="my_schema").alias("b")
    e_lookup = Table('employee', metadata, autoload_with=conn, schema="my_schema").alias("l")
    
    # Correlated subquery
    max_id_in_dept = (
        select(func.max(e_lookup.c.id))
        .where(e_base.c.dept_id == e_lookup.c.dept_id)
    ).scalar_subquery()
    
    # Delete using subquery
    delete_stmt = delete(e_base).where(e_base.c.id == max_id_in_dept)
    
    compiled = delete_stmt.compile(conn)
    conn.execute(compiled)
```

--------------------------------

### Create UNPIVOT Query with SQLAlchemy Kinetica

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates the creation of an UNPIVOT query using UnpivotSelect from sqlalchemy_kinetica.custom_commands. This process involves defining the value column, type column, and the columns to be unpivoted, often after creating a subquery with aliased columns.

```python
from sqlalchemy_kinetica.custom_commands import UnpivotSelect
from sqlalchemy import column, Table, MetaData, select

metadata = MetaData()

with engine.connect() as conn:
    customer_contact = Table('customer_contact', metadata, 
                            autoload_with=conn, schema="my_schema").alias("cc")
    
    # Create subquery with renamed columns
    subquery = select(
        customer_contact.c.name,
        customer_contact.c.home_phone.label('Home'),
        customer_contact.c.work_phone.label('Work'),
        customer_contact.c.cell_phone.label('Cell')
    )
    
    # Create UNPIVOT query
    query = (
        UnpivotSelect(
            column("name"),
            column("phone_type"),
            column("phone_number")
        )
        .select_from(subquery)
        .unpivot(
            "phone_number",           # Value column
            "phone_type",             # Type column
            ["Home", "Work", "Cell"]  # Columns to unpivot
        )
        .order_by(column("name"), column("phone_type"))
    )
    
    compiled = query.compile(conn)
    result = conn.execute(compiled)
```

--------------------------------

### UPDATE with Non-Infixed JOIN Syntax using SQLAlchemy-Kinetica

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates how to perform an UPDATE operation using a non-infixed JOIN syntax with SQLAlchemy-Kinetica. This approach utilizes the `FROM` and `WHERE` clauses to achieve the join, offering an alternative to the infixed syntax. It also uses the `KiUpdate` class.

```python
# UPDATE with FROM and WHERE (non-infixed join)
update_stmt = KiUpdate(
    employee_backup,
    from_table=employee,
    where_condition=(employee_backup.c.id == employee.c.id)
).values(
    sal=employee.c.sal,
    manager_id=employee.c.manager_id
)

compiled = update_stmt.compile(conn)
conn.execute(compiled)
```

--------------------------------

### Kinetica Insert from Select Statement

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Inserts data into a target Kinetica table by selecting records from a source table. This uses the `ki_insert` construct with the `from_select` method, allowing for conditional data population based on a query. Assumes source and target tables are defined.

```python
from sqlalchemy import select, MetaData, Table
from sqlalchemy.engine import create_engine
from sqlalchemy_kinetica.custom_commands import ki_insert

# Assume metadata and engine are defined elsewhere
# metadata = MetaData()
# engine = create_engine('kinetica://user:password@host:port/database')

# with engine.connect() as conn:
#     source = Table('employees', metadata, autoload_with=conn, schema="my_schema")
#     target = Table('employees_backup', metadata, autoload_with=conn, schema="my_schema")
    
#     # Insert from SELECT
#     insert_stmt = ki_insert(target).from_select(
#         ["id", "name", "department", "salary"],
#         select(source.c.id, source.c.name, source.c.department, source.c.salary)
#         .where(source.c.salary > 50000)
#     )
    
#     conn.execute(insert_stmt)

```

--------------------------------

### Define Shard Key for SQLAlchemy Table

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Explains how to designate a column as a shard key for a Kinetica table using SQLAlchemy. This is achieved by adding `info={'shard_key': True}` to the `Column` definition. Multiple columns can be marked as part of a composite shard key.

```python
sharded_table = Table(
    "sharded_data",
    metadata,
    Column("id", Integer, nullable=False, primary_key=True),
    Column("dept_id", Integer, nullable=False, primary_key=True, 
           info={"shard_key": True}),  # Mark as shard key
    Column("data", VARCHAR(100)),
    schema="my_schema"
)
```

--------------------------------

### Define Table with Kinetica Special Types

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to define a SQLAlchemy Table using various Kinetica-specific data types. This includes IPV4, UUID, VECTOR, DATE, TIME, DATETIME, TIMESTAMP, JSON, BLOB, BlobWKT, GEOMETRY, and others. Ensure `sqlalchemy_kinetica.kinetica_types` is imported.

```python
from sqlalchemy import Column, Integer, MetaData, Table
from sqlalchemy_kinetica.kinetica_types import *

metadata = MetaData()

all_types_table = Table(
    "various_types",
    metadata,
    Column("i", Integer, primary_key=True),
    Column("ti", TINYINT),
    Column("si", SMALLINT),
    Column("bi", BIGINT),
    Column("ub", UnsignedBigInteger),
    Column("r", REAL),
    Column("d", DOUBLE),
    Column("dc", DECIMAL(10, 4)),
    Column("s", VARCHAR(256)),
    Column("ip", IPV4),
    Column("u", UUID),
    Column("td", DATE),
    Column("tt", TIME),
    Column("dt", DATETIME),
    Column("ts", TIMESTAMP),
    Column("j", JSON),
    Column("bl", BLOB),
    Column("w", BlobWKT),
    Column("g", GEOMETRY),
    Column("v", VECTOR(10)),
    schema="my_schema"
)
```

--------------------------------

### Best Practice: Use literal_binds for Complex Queries in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Highlights the best practice of using `literal_binds=True` when compiling SQLAlchemy queries that involve arithmetic operations with numeric literals or Common Table Expressions (CTEs). This prevents potential type errors, such as `<UNKNOWN>` type issues, ensuring correct query execution.

```python
# Good - prevents <UNKNOWN> type errors
compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
result = conn.execute(compiled)
```

--------------------------------

### Perform INTERSECT Operation with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Illustrates how to find the common rows between two SELECT statements using the `intersect` function from SQLAlchemy. This returns only the rows that exist in both result sets.

```python
from sqlalchemy import intersect, select

# Assuming 'lunch' and 'dinner' tables and 'conn' are defined as in the UNION example
# lunch = Table('lunch_menu', metadata, autoload_with=conn, schema="my_schema")
# dinner = Table('dinner_menu', metadata, autoload_with=conn, schema="my_schema")

intersect_query = intersect(
    select(lunch.c.item, lunch.c.price),
    select(dinner.c.item, dinner.c.price)
)

compiled = intersect_query.compile(conn)
result = conn.execute(compiled)
```

--------------------------------

### SQL ROLLUP Aggregation in SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates the use of the ROLLUP aggregation function in SQLAlchemy to generate subtotals and a grand total for a specified column. It handles NULLs for grouping columns.

```python
from sqlalchemy import select, func, Table, case

# Assuming 'employees' table is already defined and connected via 'conn'
# employees = Table('employees', metadata, autoload_with=conn)

query = select(
    case(
        (func.grouping(employees.c.department) == 1, '<ALL DEPARTMENTS>'),
        else_=func.nvl(employees.c.department, '<UNKNOWN>')
    ).label('dept_group'),
    func.avg(employees.c.salary).label('avg_salary')
).group_by(
    func.rollup(employees.c.department)
)

compiled = query.compile(conn, compile_kwargs={"literal_binds": True})
result = conn.execute(compiled)
```

--------------------------------

### Inline Literal Values in SQLAlchemy Queries

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Demonstrates how to use `sqlalchemy.literal()` to ensure numeric literals are correctly interpreted in arithmetic expressions within SQLAlchemy queries, preventing type-related errors.

```python
from sqlalchemy import literal, select

# Assuming 'employees' is a SQLAlchemy Table object and 'salary' is a column
# This example multiplies the salary by 1.1, ensuring 1.1 is treated as a literal.
query = select(employees.c.salary * literal(1.1))
```

--------------------------------

### Perform EXCEPT Operation with SQLAlchemy

Source: https://github.com/kineticadb/sqlalchemy-kinetica/blob/master/user_guide.md

Shows how to retrieve rows from the first SELECT statement that are not present in the second SELECT statement using the `except_` function from SQLAlchemy. This is equivalent to a set difference.

```python
from sqlalchemy import except_, select

# Assuming 'lunch' and 'dinner' tables and 'conn' are defined as in the UNION example
# lunch = Table('lunch_menu', metadata, autoload_with=conn, schema="my_schema")
# dinner = Table('dinner_menu', metadata, autoload_with=conn, schema="my_schema")

except_query = except_(
    select(lunch.c.item, lunch.c.price),
    select(dinner.c.item, dinner.c.price)
)

compiled = except_query.compile(conn)
result = conn.execute(compiled)
```

--------------------------------

### CREATE TABLE AS SELECT using CreateTableAs

Source: https://context7.com/kineticadb/sqlalchemy-kinetica/llms.txt

Create new tables from query results using the CreateTableAs command. This is useful for materializing query results into a new table, potentially with options like replacing an existing table or specifying replication.

```python
from sqlalchemy import MetaData, Table, select
from sqlalchemy_kinetica.custom_commands import CreateTableAs

metadata = MetaData()

with engine.connect() as conn:
    source_table = Table('employee', metadata, autoload_with=conn, schema="my_schema")

    create_stmt = CreateTableAs(
        '"my_schema"."employee_backup"',
        select(source_table),
        prefixes=["OR REPLACE", "REPLICATED", "TEMP"]
    )

    compiled = create_stmt.compile(conn)
    conn.execute(compiled)
```