Best Queries That Save Time Tableau in 2026: For Every Budget

import os
import sys
import logging
from typing import List, Dict, Optional
from tableauhyperapi import HyperProcess, Telemetry, Connection, CreateMode, \
    HyperException, TableDefinition, TableName, SqlType, InsertTruncate, \
    escape_name, escape_string_literal

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

# Constants for Hyper API configuration
HYPER_VERSION = "3.2.0"
SUPPORTED_TABLEAU_VERSIONS = ["2024.2", "2025.1", "2026.1"]
MAX_BATCH_SIZE = 10000
QUERY_CACHE_DIR = "./hyper_query_cache"

def optimize_extract_queries(hyper_path: str, target_tables: Optional[List[str]] = None) -> Dict[str, float]:
    """
    Optimizes query execution for Tableau Hyper extracts by:
    1. Rebuilding indexes on high-cardinality columns
    2. Pruning unused columns from extracts
    3. Enabling native query plan caching (Hyper API 3.2.0+)

    Args:
        hyper_path: Path to the .hyper extract file
        target_tables: List of table names to optimize; None optimizes all tables

    Returns:
        Dict mapping table names to query time reduction percentages
    """
    if not os.path.exists(hyper_path):
        raise FileNotFoundError(f"Hyper extract not found at {hyper_path}")

    # Validate Tableau version compatibility
    if not any(ver in hyper_path for ver in SUPPORTED_TABLEAU_VERSIONS):
        logger.warning(f"Hyper file {hyper_path} may not be compatible with optimized queries for 2026 tiers")

    results = {}
    hyper_process = None
    connection = None

    try:
        # Start Hyper process with telemetry disabled for CI/CD environments
        hyper_process = HyperProcess(
            telemetry=Telemetry.DO_NOT_SEND_USAGE_DATA_TO_TABLEAU,
            parameters={"log_dir": "./hyper_logs"}
        )
        logger.info(f"Started Hyper process version {HYPER_VERSION}")

        # Open connection to Hyper extract in read-write mode
        connection = Connection(
            endpoint=hyper_process.endpoint,
            database=hyper_path,
            create_mode=CreateMode.OPEN_EXISTING
        )
        logger.info(f"Connected to Hyper extract: {hyper_path}")

        # Get list of tables to process
        tables = target_tables if target_tables else [
            table.name.unescaped for table in connection.catalog.get_tables()
        ]
        logger.info(f"Optimizing {len(tables)} tables: {tables}")

        for table_name in tables:
            esc_table = escape_name(table_name)
            # Measure baseline query time for COUNT(*) on table
            import time
            baseline_query = f"SELECT COUNT(*) FROM {esc_table}"
            start = time.perf_counter()
            connection.execute_query(baseline_query)
            baseline_time = time.perf_counter() - start
            logger.info(f"Baseline query time for {table_name}: {baseline_time:.4f}s")

            # Rebuild indexes on high-cardinality columns ( > 10% unique values)
            table_def = connection.catalog.get_table_definition(TableName("public", table_name))
            for column in table_def.columns:
                col_name = escape_name(column.name.unescaped)
                # Check cardinality
                cardinality_query = f"SELECT COUNT(DISTINCT {col_name}) FROM {esc_table}"
                distinct_count = connection.execute_query(cardinality_query).get_rows()[0][0]
                row_count = connection.execute_query(f"SELECT COUNT(*) FROM {esc_table}").get_rows()[0][0]
                if row_count == 0:
                    continue
                cardinality_ratio = distinct_count / row_count
                if cardinality_ratio > 0.1:
                    # Create index if not exists
                    index_query = f"CREATE INDEX IF NOT EXISTS idx_{table_name}_{column.name.unescaped} ON {esc_table} ({col_name})"
                    connection.execute_query(index_query)
                    logger.info(f"Created index on {table_name}.{column.name.unescaped} (cardinality ratio: {cardinality_ratio:.2f})")

            # Enable query plan caching (Hyper API 3.2.0 feature)
            cache_query = f"SET QUERY_CACHE_DIR = '{escape_string_literal(QUERY_CACHE_DIR)}'"
            connection.execute_query(cache_query)
            logger.info(f"Enabled query plan caching for {table_name}")

            # Measure optimized query time
            start = time.perf_counter()
            connection.execute_query(baseline_query)
            optimized_time = time.perf_counter() - start
            reduction = ((baseline_time - optimized_time) / baseline_time) * 100 if baseline_time > 0 else 0
            results[table_name] = reduction
            logger.info(f"Optimized query time for {table_name}: {optimized_time:.4f}s ({reduction:.2f}% reduction)")

        return results

    except HyperException as e:
        logger.error(f"Hyper API error: {e}")
        raise
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        raise
    finally:
        if connection:
            connection.close()
            logger.info("Closed Hyper connection")
        if hyper_process:
            hyper_process.close()
            logger.info("Stopped Hyper process")

if __name__ == "__main__":
    # Example usage: optimize all tables in a Hyper extract
    try:
        reduction_stats = optimize_extract_queries("./sales_data.hyper")
        logger.info(f"Overall optimization results: {reduction_stats}")
    except Exception as e:
        logger.error(f"Failed to optimize extracts: {e}")
        sys.exit(1)

import requests
import json
import time
from typing import List, Dict, Optional
import logging
from datetime import datetime, timedelta

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

# Tableau REST API constants for 2026.1
TABLEAU_API_VERSION = "3.19"  # Corresponds to Tableau 2026.1
AUTH_ENDPOINT = "/api/{version}/auth/signin"
QUERY_ENDPOINT = "/api/{version}/sites/{site_id}/queries"
SLOW_QUERY_THRESHOLD_MS = 5000  # Queries taking >5s are considered slow

class TableauQueryAuditor:
    def __init__(self, server_url: str, username: str, password: str, site_id: str = ""):
        self.server_url = server_url.rstrip("/")
        self.username = username
        self.password = password
        self.site_id = site_id
        self.auth_token = None
        self.site_id_full = None
        self.session = requests.Session()
        self.session.headers.update({"Content-Type": "application/json", "Accept": "application/json"})

    def _make_request(self, method: str, endpoint: str, data: Optional[Dict] = None, params: Optional[Dict] = None) -> Dict:
        """Wrapper for Tableau REST API requests with error handling"""
        url = f"{self.server_url}{endpoint.format(version=TABLEAU_API_VERSION, site_id=self.site_id_full or self.site_id)}"
        try:
            if method.upper() == "POST":
                response = self.session.post(url, json=data, params=params, timeout=30)
            else:
                response = self.session.get(url, params=params, timeout=30)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            logger.error(f"HTTP error {e.response.status_code} for {url}: {e.response.text}")
            raise
        except requests.exceptions.RequestException as e:
            logger.error(f"Request failed for {url}: {e}")
            raise

    def sign_in(self) -> None:
        """Authenticate to Tableau Server and retrieve auth token"""
        endpoint = AUTH_ENDPOINT
        payload = {
            "credentials": {
                "name": self.username,
                "password": self.password,
                "site": {"contentUrl": self.site_id}
            }
        }
        try:
            response = self._make_request("POST", endpoint, data=payload)
            self.auth_token = response["credentials"]["token"]
            self.site_id_full = response["credentials"]["site"]["id"]
            self.session.headers.update({"X-Tableau-Auth": self.auth_token})
            logger.info(f"Signed in to Tableau Server {self.server_url}, site ID: {self.site_id_full}")
        except Exception as e:
            logger.error(f"Sign in failed: {e}")
            raise

    def sign_out(self) -> None:
        """Sign out of Tableau Server and invalidate token"""
        if self.auth_token:
            try:
                self._make_request("POST", "/api/{version}/auth/signout")
                logger.info("Signed out of Tableau Server")
            except Exception as e:
                logger.error(f"Sign out failed: {e}")
            finally:
                self.auth_token = None
                self.site_id_full = None

    def get_slow_queries(self, days_back: int = 7) -> List[Dict]:
        """
        Retrieve all queries slower than SLOW_QUERY_THRESHOLD_MS in the last days_back days

        Args:
            days_back: Number of days to look back for slow queries

        Returns:
            List of slow query dicts with query text, duration, user, workbook
        """
        if not self.auth_token:
            raise RuntimeError("Not signed in to Tableau Server. Call sign_in() first.")

        start_time = datetime.now() - timedelta(days=days_back)
        params = {
            "filter": f"durationMs:gt:{SLOW_QUERY_THRESHOLD_MS},createdAt:gt:{start_time.strftime('%Y-%m-%dT%H:%M:%SZ')}",
            "pageSize": 100
        }
        slow_queries = []
        page = 1

        while True:
            params["pageNumber"] = page
            try:
                response = self._make_request("GET", QUERY_ENDPOINT, params=params)
                queries = response.get("queries", {}).get("query", [])
                if not queries:
                    break
                for query in queries:
                    slow_queries.append({
                        "query_id": query["id"],
                        "duration_ms": query["durationMs"],
                        "query_text": query["queryText"],
                        "user": query["user"]["name"],
                        "workbook": query.get("workbook", {}).get("name", "N/A"),
                        "created_at": query["createdAt"]
                    })
                logger.info(f"Retrieved page {page}, total slow queries so far: {len(slow_queries)}")
                if "pagination" in response and response["pagination"]["pageNumber"] >= response["pagination"]["totalPages"]:
                    break
                page += 1
            except Exception as e:
                logger.error(f"Failed to retrieve page {page}: {e}")
                break

        logger.info(f"Found {len(slow_queries)} slow queries in last {days_back} days")
        return slow_queries

    def generate_optimization_report(self, slow_queries: List[Dict]) -> Dict:
        """Generate a report with optimization suggestions for slow queries"""
        report = {
            "total_slow_queries": len(slow_queries),
            "avg_duration_ms": sum(q["duration_ms"] for q in slow_queries) / len(slow_queries) if slow_queries else 0,
            "top_offenders": {},
            "suggestions": []
        }
        workbook_counts = {}
        for q in slow_queries:
            wb = q["workbook"]
            workbook_counts[wb] = workbook_counts.get(wb, 0) + 1
        report["top_offenders"] = dict(sorted(workbook_counts.items(), key=lambda x: x[1], reverse=True)[:5])
        report["suggestions"].append("Replace nested LOD expressions with materialized extracts where possible")
        report["suggestions"].append("Enable query result caching for dashboards with >100 daily viewers")
        report["suggestions"].append("Use Hyper API 3.2.0 query plan caching for repeat datasets")
        return report

if __name__ == "__main__":
    auditor = TableauQueryAuditor(
        server_url="https://tableau-enterprise.example.com",
        username="admin@example.com",
        password="secure_password_123",
        site_id="sales"
    )
    try:
        auditor.sign_in()
        slow_queries = auditor.get_slow_queries(days_back=7)
        report = auditor.generate_optimization_report(slow_queries)
        print(json.dumps(report, indent=2))
        with open(f"slow_query_report_{datetime.now().strftime('%Y%m%d')}.json", "w") as f:
            json.dump(report, f, indent=2)
        logger.info("Saved optimization report to file")
    except Exception as e:
        logger.error(f"Audit failed: {e}")
    finally:
        auditor.sign_out()

import psycopg2
import time
from typing import List, Dict, Tuple
import logging
from dataclasses import dataclass

# Configure logging
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

@dataclass
class QueryBenchmark:
    name: str
    query: str
    avg_time_ms: float
    row_count: int

class TableauQueryBenchmarker:
    def __init__(self, db_params: Dict[str, str]):
        """
        Initialize benchmarker with PostgreSQL connection params

        Args:
            db_params: Dict with keys host, port, dbname, user, password
        """
        self.db_params = db_params
        self.conn = None

    def connect(self) -> None:
        """Establish connection to PostgreSQL database"""
        try:
            self.conn = psycopg2.connect(**self.db_params)
            self.conn.autocommit = True
            logger.info(f"Connected to PostgreSQL database {self.db_params['dbname']} on {self.db_params['host']}")
        except psycopg2.Error as e:
            logger.error(f"PostgreSQL connection failed: {e}")
            raise

    def disconnect(self) -> None:
        """Close database connection"""
        if self.conn:
            self.conn.close()
            logger.info("Disconnected from PostgreSQL")

    def run_benchmark(self, query: str, iterations: int = 10) -> Tuple[float, int]:
        """
        Run a query multiple times and return average time and row count

        Args:
            query: SQL query to benchmark
            iterations: Number of times to run the query

        Returns:
            Tuple of (average time in ms, row count)
        """
        if not self.conn:
            raise RuntimeError("Not connected to database. Call connect() first.")

        total_time_ms = 0.0
        row_count = 0

        for i in range(iterations):
            try:
                start = time.perf_counter()
                with self.conn.cursor() as cur:
                    cur.execute(query)
                    row_count = cur.rowcount if cur.rowcount >= 0 else len(cur.fetchall())
                elapsed_ms = (time.perf_counter() - start) * 1000
                total_time_ms += elapsed_ms
                logger.debug(f"Iteration {i+1}: {elapsed_ms:.2f}ms, {row_count} rows")
            except psycopg2.Error as e:
                logger.error(f"Query failed on iteration {i+1}: {e}")
                raise

        avg_time_ms = total_time_ms / iterations
        return avg_time_ms, row_count

    def compare_queries(self, benchmarks: List[QueryBenchmark], iterations: int = 10) -> Dict:
        """
        Compare multiple queries and generate a report

        Args:
            benchmarks: List of QueryBenchmark objects to compare
            iterations: Number of iterations per query

        Returns:
            Dict with comparison results
        """
        results = []
        for bench in benchmarks:
            logger.info(f"Benchmarking query: {bench.name}")
            avg_time, row_count = self.run_benchmark(bench.query, iterations)
            results.append({
                "name": bench.name,
                "avg_time_ms": avg_time,
                "row_count": row_count,
                "iterations": iterations
            })
            logger.info(f"Query {bench.name}: {avg_time:.2f}ms avg, {row_count} rows")

        if len(results) >= 2:
            baseline = results[0]["avg_time_ms"]
            for res in results[1:]:
                res["speedup_vs_baseline"] = baseline / res["avg_time_ms"] if res["avg_time_ms"] > 0 else 0
        return {"results": results, "baseline_query": results[0]["name"] if results else None}

if __name__ == "__main__":
    db_params = {
        "host": "tableau-data.example.com",
        "port": 5432,
        "dbname": "sales_db",
        "user": "tableau_user",
        "password": "secure_db_password"
    }

    benchmarks = [
        QueryBenchmark(
            name="Unoptimized: Nested subqueries for monthly sales per region",
            query="""
                SELECT r.region_name, DATE_TRUNC('month', o.order_date) AS month,
                    SUM(o.total_amount) AS total_sales,
                    (SELECT COUNT(*) FROM orders o2 WHERE o2.region_id = r.region_id 
                     AND DATE_TRUNC('month', o2.order_date) = DATE_TRUNC('month', o.order_date)) AS order_count
                FROM regions r JOIN orders o ON r.region_id = o.region_id
                WHERE o.order_date >= '2025-01-01'
                GROUP BY r.region_name, DATE_TRUNC('month', o.order_date), r.region_id
                ORDER BY month DESC, total_sales DESC
            """
        ),
        QueryBenchmark(
            name="Optimized: CTE + window functions for monthly sales per region",
            query="""
                WITH monthly_orders AS (
                    SELECT r.region_name, DATE_TRUNC('month', o.order_date) AS month,
                        o.total_amount, o.region_id,
                        COUNT(*) OVER (PARTITION BY o.region_id, DATE_TRUNC('month', o.order_date)) AS order_count
                    FROM regions r JOIN orders o ON r.region_id = o.region_id
                    WHERE o.order_date >= '2025-01-01'
                )
                SELECT region_name, month, SUM(total_amount) AS total_sales, MAX(order_count) AS order_count
                FROM monthly_orders GROUP BY region_name, month ORDER BY month DESC, total_sales DESC
            """
        ),
        QueryBenchmark(
            name="Optimized: Materialized view with pre-aggregated monthly sales",
            query="""
                SELECT region_name, month, total_sales, order_count
                FROM mv_monthly_sales_by_region WHERE month >= '2025-01-01'
                ORDER BY month DESC, total_sales DESC
            """
        )
    ]

    benchmarker = TableauQueryBenchmarker(db_params)
    try:
        benchmarker.connect()
        comparison = benchmarker.compare_queries(benchmarks, iterations=15)
        print("\n=== Query Benchmark Results ===")
        for res in comparison["results"]:
            print(f"Query: {res['name']}")
            print(f"  Avg Time: {res['avg_time_ms']:.2f}ms")
            print(f"  Row Count: {res['row_count']}")
            if "speedup_vs_baseline" in res:
                print(f"  Speedup vs Baseline: {res['speedup_vs_baseline']:.2f}x")
            print()
        import json
        with open("query_benchmark_results.json", "w") as f:
            json.dump(comparison, f, indent=2)
        logger.info("Saved benchmark results to query_benchmark_results.json")
    except Exception as e:
        logger.error(f"Benchmarking failed: {e}")
    finally:
        benchmarker.disconnect()

# Enable query plan caching in Hyper API 3.2.0+
cache_dir = "./hyper_query_cache"
connection.execute_query(f"SET QUERY_CACHE_DIR = '{cache_dir}'")
logger.info(f"Enabled query plan caching at {cache_dir}")

# Materialize LOD calculation in Hyper extract (Hyper SQL)
materialize_query = """
    ALTER TABLE sales_data ADD COLUMN regional_product_sales DOUBLE PRECISION;
    UPDATE sales_data SET regional_product_sales = (SELECT SUM(s2.sales) FROM sales_data s2 
        WHERE s2.region_id = sales_data.region_id AND s2.product_id = sales_data.product_id)
"""
connection.execute_query(materialize_query)

# Retrieve slow queries from Tableau REST API
slow_queries = auditor.get_slow_queries(days_back=7)
print(f"Found {len(slow_queries)} slow queries this week")