package sqls import ( "context" "fmt" "os" "sort" "strconv" "strings" "time" "unicode" "github.com/pingcap/log" o11yv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/utils/duckdb" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/utils/metacache" "go.uber.org/zap" "sigs.k8s.io/controller-runtime/pkg/client" ) type DuckDBAdapter struct { vendor string region string mataCache metacache.MetaCache duckdb duckdb.QueryDuckDBFetcherInterface store storage.Storage kubeCli client.Client } const sqlsDataSourceV2 = "v2" func NewQueryDuckDBAdapter(vendor, region string, store storage.Storage, kubeCli client.Client) *DuckDBAdapter { adapter := &DuckDBAdapter{ vendor: vendor, region: region, mataCache: metacache.NewMetaCache(kubeCli, store), kubeCli: kubeCli, } newDuckDB, err := duckdb.NewQueryDuckDBFetcherWithPool(context.Background(), vendor, region) if err != nil { log.Error("Failed to initialize DuckDB", zap.Error(err)) panic(err) } adapter.duckdb = newDuckDB adapter.store = store // Start the warmup process in a goroutine // no need to warmup for pooled connection // go func() { // err := adapter.warmup(context.Background(), newDuckDB) // if err != nil { // log.Error("Failed to warmup DuckDB", zap.Error(err)) // } // adapter.startWarmupRefreshLoop(context.Background()) // }() return adapter } //nolint:unused // Kept for potential future use func (s *DuckDBAdapter) startWarmupRefreshLoop(ctx context.Context) { refreshInterval := 2 * time.Hour ticker := time.NewTicker(refreshInterval) defer ticker.Stop() for { select { case <-ticker.C: log.Debug("Refreshing DuckDB") err := s.RenewAndWarmupDuckDB(ctx) if err != nil { log.Error("Failed to re-initialize DuckDB", zap.Error(err)) continue } else { log.Info("DuckDB re-initialized") } } } } func (s *DuckDBAdapter) RenewAndWarmupDuckDB(ctx context.Context) error { log.Debug("Initializing DuckDB for sqls query adapter") defer log.Debug("DuckDB initialization completed") newDuckDB, err := duckdb.NewQueryDuckDBFetcherWithPool(context.Background(), s.vendor, s.region) if err != nil { log.Error("Failed to reinitialize DuckDB", zap.Error(err)) return err } err = s.warmup(ctx, newDuckDB) if err != nil { log.Error("Failed to warmup DuckDB", zap.Error(err)) return err } originalDuckDB := s.duckdb s.duckdb = newDuckDB if originalDuckDB != nil { go func() { time.Sleep(time.Minute * 5) if err := originalDuckDB.Close(context.Background()); err != nil { log.Error("Failed to close original DuckDB", zap.Error(err)) } else { log.Info("Original DuckDB closed successfully") } }() } return nil } const warmupQuery = "SELECT count(*) as count FROM %s WHERE date >= '%s';" func (s *DuckDBAdapter) warmup(ctx context.Context, db duckdb.QueryDuckDBFetcherInterface) error { log.Debug("Starting warmup for DuckDB sqls query adapter") defer log.Debug("DuckDB warmup completed") date := time.Now().Format("20060102") clusterTables, err := s.getClusterTables(ctx) if err != nil { log.Error("Failed to get cluster tables", zap.Error(err)) } type warmupResult struct { Count int64 `db:"count"` } // Execute the query using a custom query function that handles the custom result type var warmupResults []warmupResult for alias, table := range clusterTables { err := db.SelectSlowlogInDLWithoutFilterPushdown(ctx, &warmupResults, alias, table, fmt.Sprintf(warmupQuery, alias, date)) if err != nil { log.Error("warmup failed", zap.String("tableAlias", alias), zap.Error(err)) continue } } return nil } func tableAlias(clusterID string, table string) string { tableName := fmt.Sprintf("%s_%s", table, clusterID) return strings.ReplaceAll(tableName, "-", "_") } // buildListSQL constructs the SQL query for the List operation. // //nolint:unused // Stub function, kept for potential future use func buildListSQL(params *ListTableDataParams, slowlogTablePath string) (string, string, error) { fullQuery := fmt.Sprintf("SELECT 1;") return fullQuery, "", nil } var fieldsMap = map[string]string{ "summary_begin_time": "min(epoch(SUMMARY_BEGIN_TIME)) as summary_begin_time", "summary_end_time": "max(epoch(SUMMARY_END_TIME)) as summary_end_time", "digest_text": "any_value(digest_text) as digest_text", "digest": "digest", "exec_count": "sum(exec_count) as exec_count", "stmt_type": "any_value(stmt_type) as stmt_type", "sum_errors": "sum(sum_errors) as sum_errors", "sum_warnings": "sum(sum_warnings) as sum_warnings", "sum_latency": "sum(sum_latency) as sum_latency", "max_latency": "max(max_latency) as max_latency", "min_latency": "min(min_latency) as min_latency", "avg_latency": "sum(sum_latency)/sum(exec_count) as avg_latency", "avg_parse_latency": "sum(avg_parse_latency*exec_count)/sum(exec_count) as avg_parse_latency", "max_parse_latency": "max(max_parse_latency) as max_parse_latency", "avg_compile_latency": "sum(avg_compile_latency*exec_count)/sum(exec_count) as avg_compile_latency", "max_compile_latency": "max(max_compile_latency) as max_compile_latency", "sum_cop_task_num": "sum(sum_cop_task_num) as sum_cop_task_num", "max_cop_process_time": "max(max_cop_process_time) as max_cop_process_time", "max_cop_wait_time": "max(max_cop_wait_time) as max_cop_wait_time", "avg_process_time": "sum(avg_process_time*exec_count)/sum(exec_count) as avg_process_time", "max_process_time": "max(max_process_time) as max_process_time", "avg_wait_time": "sum(avg_wait_time*exec_count)/sum(exec_count) as avg_wait_time", "max_wait_time": "max(max_wait_time) as max_wait_time", "avg_backoff_time": "sum(avg_backoff_time*exec_count)/sum(exec_count) as avg_backoff_time", "max_backoff_time": "max(max_backoff_time) as max_backoff_time", "avg_total_keys": "sum(avg_total_keys*exec_count)/sum(exec_count) as avg_total_keys", "max_total_keys": "max(max_total_keys) as max_total_keys", "avg_processed_keys": "sum(avg_processed_keys*exec_count)/sum(exec_count) as avg_processed_keys", "max_processed_keys": "max(max_processed_keys) as max_processed_keys", "avg_prewrite_time": "sum(avg_prewrite_time*exec_count)/sum(exec_count) as avg_prewrite_time", "max_prewrite_time": "max(max_prewrite_time) as max_prewrite_time", "avg_commit_time": "sum(avg_commit_time*exec_count)/sum(exec_count) as avg_commit_time", "max_commit_time": "max(max_commit_time) as max_commit_time", "avg_get_commit_ts_time": "sum(avg_get_commit_ts_time*exec_count)/sum(exec_count) as avg_get_commit_ts_time", "max_get_commit_ts_time": "max(max_get_commit_ts_time) as max_get_commit_ts_time", "avg_commit_backoff_time": "sum(avg_commit_backoff_time*exec_count)/sum(exec_count) as avg_commit_backoff_time", "max_commit_backoff_time": "max(max_commit_backoff_time) as max_commit_backoff_time", "avg_resolve_lock_time": "sum(avg_resolve_lock_time*exec_count)/sum(exec_count) as avg_resolve_lock_time", "max_resolve_lock_time": "max(max_resolve_lock_time) as max_resolve_lock_time", "avg_local_latch_wait_time": "sum(avg_local_latch_wait_time*exec_count)/sum(exec_count) as avg_local_latch_wait_time", "max_local_latch_wait_time": "max(max_local_latch_wait_time) as max_local_latch_wait_time", "avg_write_keys": "sum(avg_write_keys*exec_count)/sum(exec_count) as avg_write_keys", "max_write_keys": "max(max_write_keys) as max_write_keys", "avg_write_size": "sum(avg_write_size*exec_count)/sum(exec_count) as avg_write_size", "max_write_size": "max(max_write_size) as max_write_size", "avg_prewrite_regions": "sum(avg_prewrite_regions*exec_count)/sum(exec_count) as avg_prewrite_regions", "max_prewrite_regions": "max(max_prewrite_regions) as max_prewrite_regions", "avg_txn_retry": "sum(avg_txn_retry*exec_count)/sum(exec_count) as avg_txn_retry", "max_txn_retry": "max(max_txn_retry) as max_txn_retry", "sum_backoff_times": "sum(sum_backoff_times) as sum_backoff_times", "avg_mem": "sum(avg_mem*exec_count)/sum(exec_count) as avg_mem", "max_mem": "max(max_mem) as max_mem", "avg_disk": "sum(avg_disk*exec_count)/sum(exec_count) as avg_disk", "max_disk": "max(max_disk) as max_disk", "avg_affected_rows": "sum(avg_affected_rows*exec_count)/sum(exec_count) as avg_affected_rows", "first_seen": "min(epoch(first_seen)) as first_seen", "last_seen": "max(epoch(last_seen)) as last_seen", "sample_user": "any_value(sample_user) as sample_user", "user": "any_value(sample_user) as user", "query_sample_text": "any_value(QUERY_SAMPLE_TEXT) as query_sample_text", "prev_sample_text": "any_value(prev_sample_text) as prev_sample_text", "schema_name": "any_value(schema_name) as schema_name", "table_names": "any_value(table_names) as table_names", "index_names": "any_value(index_names) as index_names", "plan_count": "count(distinct plan_digest) as plan_count", "plan": "any_value(plan) as plan", "plan_digest": "any_value(plan_digest) as plan_digest", "binary_plan": "any_value(binary_plan) as binary_plan", "plan_hint": "any_value(plan_hint) as plan_hint", "plan_in_binding": "any_value(plan_in_binding) as plan_in_binding", "max_rocksdb_delete_skipped_count": "max(max_rocksdb_delete_skipped_count) as max_rocksdb_delete_skipped_count", "avg_rocksdb_delete_skipped_count": "sum(avg_rocksdb_delete_skipped_count*exec_count)/sum(exec_count) as avg_rocksdb_delete_skipped_count", "max_rocksdb_key_skipped_count": "max(max_rocksdb_key_skipped_count) as max_rocksdb_key_skipped_count", "avg_rocksdb_key_skipped_count": "sum(avg_rocksdb_key_skipped_count*exec_count)/sum(exec_count) as avg_rocksdb_key_skipped_count", "max_rocksdb_block_cache_hit_count": "max(max_rocksdb_block_cache_hit_count) as max_rocksdb_block_cache_hit_count", "avg_rocksdb_block_cache_hit_count": "sum(avg_rocksdb_block_cache_hit_count*exec_count)/sum(exec_count) as avg_rocksdb_block_cache_hit_count", "max_rocksdb_block_read_count": "max(max_rocksdb_block_read_count) as max_rocksdb_block_read_count", "avg_rocksdb_block_read_count": "sum(avg_rocksdb_block_read_count*exec_count)/sum(exec_count) as avg_rocksdb_block_read_count", "max_rocksdb_block_read_byte": "max(max_rocksdb_block_read_byte) as max_rocksdb_block_read_byte", "avg_rocksdb_block_read_byte": "sum(avg_rocksdb_block_read_byte*exec_count)/sum(exec_count) as avg_rocksdb_block_read_byte", "sum_ru": "sum(AVG_REQUEST_UNIT_READ*exec_count) + sum(AVG_REQUEST_UNIT_WRITE*exec_count) as sum_ru", "sum_rru": "sum(AVG_REQUEST_UNIT_READ*exec_count) as sum_rru", "sum_wru": "sum(AVG_REQUEST_UNIT_WRITE*exec_count) as sum_wru", "sum_ru_wait_duration": "sum(AVG_QUEUED_RC_TIME*exec_count) as sum_ru_wait_duration", "avg_ru": "(sum(AVG_REQUEST_UNIT_READ*exec_count) + sum(AVG_REQUEST_UNIT_WRITE*exec_count))/sum(exec_count) as avg_ru", "avg_rru": "sum(AVG_REQUEST_UNIT_READ*exec_count)/sum(exec_count) as avg_rru", "avg_wru": "sum(AVG_REQUEST_UNIT_WRITE*exec_count)/sum(exec_count) as avg_wru", "max_rru": "max(MAX_REQUEST_UNIT_READ) as max_rru", "max_wru": "max(MAX_REQUEST_UNIT_WRITE) as max_wru", "max_ru_wait_duration": "max(MAX_QUEUED_RC_TIME) as max_ru_wait_duration", "max_ru": "max(MAX_REQUEST_UNIT_READ + MAX_REQUEST_UNIT_WRITE) as max_ru", "sum_ru_v2": "sum(COALESCE(AVG_REQUEST_UNIT_V2,0)*exec_count) as sum_ru_v2", "avg_ru_v2": "sum(COALESCE(AVG_REQUEST_UNIT_V2,0)*exec_count)/sum(exec_count) as avg_ru_v2", "max_ru_v2": "max(COALESCE(MAX_REQUEST_UNIT_V2,0)) as max_ru_v2", "sum_tidb_cpu": "sum(sum_tidb_cpu) as sum_tidb_cpu", "avg_tidb_cpu": "sum(sum_tidb_cpu)/sum(exec_count) as avg_tidb_cpu", "sum_tikv_cpu": "sum(sum_tikv_cpu) as sum_tikv_cpu", "avg_tikv_cpu": "sum(sum_tikv_cpu)/sum(exec_count) as avg_tikv_cpu", "plan_cache_hits": "sum(plan_cache_hits) as plan_cache_hits", "plan_in_cache": "any_value(plan_in_cache) as plan_in_cache", "plan_cache_unqualified_count": "sum(plan_cache_unqualified_count) as plan_cache_unqualified_count", "plan_cache_unqualified_last_reason": "any_value(plan_cache_unqualified_last_reason) as plan_cache_unqualified_last_reason", "sum_ia_read_segment_count": "sum(sum_ia_read_segment_count) as sum_ia_read_segment_count", "avg_ia_read_segment_count": "sum(sum_ia_read_segment_count)/sum(exec_count) as avg_ia_read_segment_count", "max_ia_read_segment_count": "max(max_ia_read_segment_count) as max_ia_read_segment_count", "sum_ia_remote_read_segment_size": "sum(sum_ia_remote_read_segment_size) as sum_ia_remote_read_segment_size", "avg_ia_remote_read_segment_size": "sum(sum_ia_remote_read_segment_size)/sum(exec_count) as avg_ia_remote_read_segment_size", "max_ia_remote_read_segment_size": "max(max_ia_remote_read_segment_size) as max_ia_remote_read_segment_size", "sum_ia_remote_read_segment_wait_time": "sum(sum_ia_remote_read_segment_wait_time) as sum_ia_remote_read_segment_wait_time", "avg_ia_remote_read_segment_wait_time": "sum(sum_ia_remote_read_segment_wait_time)/sum(exec_count) as avg_ia_remote_read_segment_wait_time", "max_ia_remote_read_segment_wait_time": "max(max_ia_remote_read_segment_wait_time) as max_ia_remote_read_segment_wait_time", } var fieldsMapV2 = map[string]string{ "summary_begin_time": "min(\"begin\") as summary_begin_time", "summary_end_time": "max(\"end\") as summary_end_time", "digest_text": "any_value(normalized_sql) as digest_text", "query_sample_text": "any_value(sample_sql) as query_sample_text", "prev_sample_text": "any_value(prev_sql) as prev_sample_text", "plan": "any_value(sample_plan) as plan", "binary_plan": "any_value(sample_binary_plan) as binary_plan", "sum_ru": "sum(sum_rru) + sum(sum_wru) as sum_ru", "sum_rru": "sum(sum_rru) as sum_rru", "sum_wru": "sum(sum_wru) as sum_wru", "avg_ru": "(sum(sum_rru) + sum(sum_wru))/sum(exec_count) as avg_ru", "avg_rru": "sum(sum_rru)/sum(exec_count) as avg_rru", "avg_wru": "sum(sum_wru)/sum(exec_count) as avg_wru", "max_rru": "max(max_rru) as max_rru", "max_wru": "max(max_wru) as max_wru", "max_ru": "max(max_rru + max_wru) as max_ru", "sum_ru_v2": "sum(sum_ruv2) as sum_ru_v2", "avg_ru_v2": "sum(sum_ruv2)/sum(exec_count) as avg_ru_v2", "max_ru_v2": "max(max_ruv2) as max_ru_v2", "avg_parse_latency": "sum(sum_parse_latency)/sum(exec_count) as avg_parse_latency", "avg_compile_latency": "sum(sum_compile_latency)/sum(exec_count) as avg_compile_latency", "sum_cop_task_num": "sum(sum_num_cop_tasks) as sum_cop_task_num", "avg_process_time": "sum(sum_process_time)/sum(exec_count) as avg_process_time", "avg_wait_time": "sum(sum_wait_time)/sum(exec_count) as avg_wait_time", "avg_backoff_time": "sum(sum_backoff_time)/sum(exec_count) as avg_backoff_time", "avg_total_keys": "sum(sum_total_keys)/sum(exec_count) as avg_total_keys", "avg_processed_keys": "sum(sum_processed_keys)/sum(exec_count) as avg_processed_keys", "avg_prewrite_time": "sum(sum_prewrite_time)/sum(exec_count) as avg_prewrite_time", "avg_commit_time": "sum(sum_commit_time)/sum(exec_count) as avg_commit_time", "avg_get_commit_ts_time": "sum(sum_get_commit_ts_time)/sum(exec_count) as avg_get_commit_ts_time", "avg_commit_backoff_time": "sum(sum_commit_backoff_time)/sum(exec_count) as avg_commit_backoff_time", "avg_resolve_lock_time": "sum(sum_resolve_lock_time)/sum(exec_count) as avg_resolve_lock_time", "avg_local_latch_wait_time": "sum(sum_local_latch_time)/sum(exec_count) as avg_local_latch_wait_time", "avg_write_keys": "sum(sum_write_keys)/sum(exec_count) as avg_write_keys", "avg_write_size": "sum(sum_write_size)/sum(exec_count) as avg_write_size", "avg_prewrite_regions": "sum(sum_prewrite_region_num)/sum(exec_count) as avg_prewrite_regions", "avg_txn_retry": "sum(sum_txn_retry)/sum(exec_count) as avg_txn_retry", "avg_mem": "sum(sum_mem)/sum(exec_count) as avg_mem", "max_mem": "max(max_mem) as max_mem", "avg_disk": "sum(sum_disk)/sum(exec_count) as avg_disk", "max_disk": "max(max_disk) as max_disk", "avg_affected_rows": "sum(sum_affected_rows)/sum(exec_count) as avg_affected_rows", "first_seen": "min(first_seen) as first_seen", "last_seen": "max(last_seen) as last_seen", "sample_user": "any_value(auth_users) as sample_user", // user must be quoted/mapped: bare "user" is DuckDB's current_user() and returns "duckdb". "user": "any_value(list_element(coalesce(json_keys(auth_users), []), 1)) as user", // v2 parquet stores sum_rocksdb_* / max_rocksdb_* (not avg_rocksdb_*). // Derive avg from sum so plan/detail schema filtering keeps these API fields. "avg_rocksdb_delete_skipped_count": "sum(sum_rocksdb_delete_skipped_count)/sum(exec_count) as avg_rocksdb_delete_skipped_count", "avg_rocksdb_key_skipped_count": "sum(sum_rocksdb_key_skipped_count)/sum(exec_count) as avg_rocksdb_key_skipped_count", "avg_rocksdb_block_cache_hit_count": "sum(sum_rocksdb_block_cache_hit_count)/sum(exec_count) as avg_rocksdb_block_cache_hit_count", "avg_rocksdb_block_read_count": "sum(sum_rocksdb_block_read_count)/sum(exec_count) as avg_rocksdb_block_read_count", "avg_rocksdb_block_read_byte": "sum(sum_rocksdb_block_read_byte)/sum(exec_count) as avg_rocksdb_block_read_byte", "sum_tidb_cpu": "sum(sum_tidb_cpu) as sum_tidb_cpu", "avg_tidb_cpu": "sum(sum_tidb_cpu)/sum(exec_count) as avg_tidb_cpu", "sum_tikv_cpu": "sum(sum_tikv_cpu) as sum_tikv_cpu", "avg_tikv_cpu": "sum(sum_tikv_cpu)/sum(exec_count) as avg_tikv_cpu", "plan_cache_hits": "sum(plan_cache_hits) as plan_cache_hits", "plan_in_cache": "any_value(plan_in_cache) as plan_in_cache", "plan_cache_unqualified_count": "sum(plan_cache_unqualified_count) as plan_cache_unqualified_count", "plan_cache_unqualified_last_reason": "any_value(plan_cache_unqualified_last_reason) as plan_cache_unqualified_last_reason", "sum_ia_read_segment_count": "sum(sum_ia_read_segment_count) as sum_ia_read_segment_count", "avg_ia_read_segment_count": "sum(sum_ia_read_segment_count)/sum(exec_count) as avg_ia_read_segment_count", "max_ia_read_segment_count": "max(max_ia_read_segment_count) as max_ia_read_segment_count", "sum_ia_remote_read_segment_size": "sum(sum_ia_remote_read_segment_size) as sum_ia_remote_read_segment_size", "avg_ia_remote_read_segment_size": "sum(sum_ia_remote_read_segment_size)/sum(exec_count) as avg_ia_remote_read_segment_size", "max_ia_remote_read_segment_size": "max(max_ia_remote_read_segment_size) as max_ia_remote_read_segment_size", "sum_ia_remote_read_segment_wait_time": "sum(sum_ia_remote_read_segment_wait_time) as sum_ia_remote_read_segment_wait_time", "avg_ia_remote_read_segment_wait_time": "sum(sum_ia_remote_read_segment_wait_time)/sum(exec_count) as avg_ia_remote_read_segment_wait_time", "max_ia_remote_read_segment_wait_time": "max(max_ia_remote_read_segment_wait_time) as max_ia_remote_read_segment_wait_time", } var toRemoveFieldMap = map[string]bool{ "connection_id": true, "digest": true, "timestamp": true, "query_time": true, "memory_max": true, } func convertDeltalakeFields(fields string) string { fieldList := strings.Split(fields, ",") selectFields := []string{ "id", "conn_id", "digest", "time", "query_time", "mem_max", } for _, field := range fieldList { if _, ok := toRemoveFieldMap[field]; ok { continue } if field == "ru" { selectFields = append(selectFields, "(request_unit_read + request_unit_write) AS ru") } else { selectFields = append(selectFields, fieldAdapt(field, "")) } } return strings.TrimSuffix(strings.Join(selectFields, ", "), ", ") } func sqlsFieldMap(dataSource string) map[string]string { if isSQLsDataSourceV2(dataSource) { merged := make(map[string]string, len(fieldsMap)+len(fieldsMapV2)) for k, v := range fieldsMap { merged[k] = v } for k, v := range fieldsMapV2 { merged[k] = v } return merged } return fieldsMap } func fieldAdapt(field, dataSource string) string { if mappedField, ok := sqlsFieldMap(dataSource)[field]; ok { return mappedField } else { return field } } // AvailableFields is not implemented for DuckDBAdapter. func (s *DuckDBAdapter) AvailableFields(ctx context.Context) ([]string, error) { //TODO implement me panic("AvailableFields not implemented for SlowQueryDuckDBAdapter") } // getSQLsBaseURL constructs the base path for DuckDB to read the slow log data table // based on the cloud provider. func (s *DuckDBAdapter) getSQLsBaseURL(tenantID, o11yClusterID, env, provider, region string, sharedTier, nextGen bool) string { var scheme string // The bucket name is determined by the `bucketName` function, which you confirmed // works for all providers with your current naming convention. bucket := storage.BucketName(env, region, sharedTier, nextGen) // Determine the URI scheme based on the provider switch strings.ToLower(provider) { case "aws": scheme = fmt.Sprintf("s3://%s", bucket) case "gcp": scheme = fmt.Sprintf("gs://%s", bucket) // Google Cloud Storage uses gs:// prefix case "azure": account := os.Getenv("AZURE_STORAGE_ACCOUNT") scheme = fmt.Sprintf("abfss://%s@%s.dfs.core.windows.net", bucket, account) // Azure Blob Storage uses abfss:// prefix case "alicloud": scheme = fmt.Sprintf("s3://%s", bucket) case "local": scheme = "./" default: // Fallback or error if provider is not recognized. log.Error("Unsupported cloud provider for slow log base URL", zap.String("provider", provider)) return "" // Return empty string to indicate an error, which should be handled by the caller } // Construct the full path return fmt.Sprintf("%s/deltalake/%s/%s", scheme, tenantID, o11yClusterID) } func isSQLsDataSourceV2(dataSource string) bool { return strings.EqualFold(strings.TrimSpace(dataSource), sqlsDataSourceV2) } func (s *DuckDBAdapter) getSQLsV2Path(tenantID, clusterID, env, provider, region string, sharedTier, nextGen bool) string { var scheme string bucket := storage.BucketName(env, region, sharedTier, nextGen) switch strings.ToLower(provider) { case "aws": scheme = fmt.Sprintf("s3://%s", bucket) case "gcp": scheme = fmt.Sprintf("gs://%s", bucket) case "azure": account := os.Getenv("AZURE_STORAGE_ACCOUNT") scheme = fmt.Sprintf("abfss://%s@%s.dfs.core.windows.net", bucket, account) case "alicloud": scheme = fmt.Sprintf("s3://%s", bucket) case "local": scheme = "." default: log.Error("Unsupported cloud provider for sqls v2 path", zap.String("provider", provider)) return "" } return fmt.Sprintf("%s/deltalake/org=%s/cluster=%s/type=sqlstatement/component=sqlstatement", scheme, tenantID, clusterID) } func buildSQLsTablePath(basePath, tableName, dataSource string) string { if isSQLsDataSourceV2(dataSource) { return basePath } return fmt.Sprintf("%s/%s", basePath, tableName) } var ( labelBizType = "observability.pingcap.com/bizType" labelBizTypeValue = "prod-tidbcloud-nextgen" labelTenantID = "observability.pingcap.com/tenantID" labelApplicationID = "observability.pingcap.com/applicationID" ) func (s *DuckDBAdapter) getClusterTables(ctx context.Context) (map[string]string, error) { // For local vendor, return a fixed list of tables if s.vendor == "local" { return map[string]string{ // "local_cluster_1": "./deltalake/tenant1/cluster1/slowlogs", // "local_cluster_2": "./deltalake/tenant2/cluster2/slowlogs", }, nil } tablesMap := make(map[string]string) clusters := &o11yv1.ClusterList{} if err := s.kubeCli.List(ctx, clusters, client.MatchingLabels{ labelBizType: labelBizTypeValue, }); err != nil { return nil, err } for _, cluster := range clusters.Items { if cluster.Labels[labelBizType] != labelBizTypeValue { continue } applicationID := cluster.Labels[labelApplicationID] if applicationID == "dataapp" || applicationID == "k8s-infra" || applicationID == "seed" || applicationID == "base" { continue } if isPureDigits(applicationID) { tableUrl := s.getSQLsBaseURL(cluster.Labels[labelTenantID], cluster.Name, "prod", s.vendor, s.region, false, false, ) tablesMap[tableAlias(applicationID, "persisted_statements_summary")] = tableUrl + "/persisted_statements_summary" } } return tablesMap, nil } func isPureDigits(s string) bool { if s == "" { return false } for _, r := range s { if !unicode.IsDigit(r) { return false } } return true } // Helper function to parse string to uint64, returning 0 if invalid func parseStringToUint64(s string) uint64 { val, err := strconv.ParseUint(s, 10, 64) if err != nil { return 0 // Return 0 or handle error as appropriate for your application } return val } // Helper function to convert int64 to uint64, returning 0 if negative func parseInt64ToUint64(i int64) uint64 { if i < 0 { return 0 // Or handle as an error/specific default if negative values are not expected } return uint64(i) } // List lists slow logs which match the conditions described in ListParams. func (s *DuckDBAdapter) ListTables(ctx context.Context, params *ListTableParams) ([]string, error) { log.Info("QueryDuckDBAdapter ListTables called", zap.String("provider", params.Provider), zap.String("region", params.Region), zap.String("tenant", params.TenantID), zap.String("project", params.ProjectID), zap.String("cluster", params.ClusterID), zap.Bool("sharedTier", params.SharedTier), ) if isSQLsDataSourceV2(params.DataSource) { return []string{STATEMENTS_TABLE_NAME}, nil } metaInfo, err := s.mataCache.GetMetaInfo(ctx, params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.NextGen, params.Byoc) if err != nil { log.Error("GetMetaInfo failed", zap.Error(err)) return nil, err } if metaInfo == nil { return nil, fmt.Errorf("meta info not found for tenant %s and cluster %s", params.TenantID, params.ClusterID) } bucket := storage.BucketName(params.Env, params.Region, params.SharedTier, params.NextGen) prefix := fmt.Sprintf("deltalake/%s/%s/", params.TenantID, metaInfo.O11yID) dirs, err := s.store.ObjectsDelimiter(ctx, bucket, prefix) if err != nil { return nil, fmt.Errorf("failed to list directories: %w", err) } if len(dirs) == 0 { return nil, fmt.Errorf("no directories found for tenant %s and cluster %s", params.TenantID, params.ClusterID) } dirSet := make(map[string]struct{}) for _, dir := range dirs { if dir == "" { continue } dir = strings.TrimPrefix(dir, prefix) parts := strings.SplitN(dir, "/", 2) if len(parts) > 0 && parts[0] != "" { dirSet[parts[0]] = struct{}{} } } tables := make([]string, 0, len(dirSet)) for dir := range dirSet { tables = append(tables, dir) } log.Info("List tables finished", zap.Int("count", len(tables))) return tables, nil } func (s *DuckDBAdapter) ListTableFields(ctx context.Context, params *ListTableFieldsParams) ([]FieldInfo, error) { log.Info("QueryDuckDBAdapter ListFields called", zap.String("provider", params.Provider), zap.String("region", params.Region), zap.String("tenant", params.TenantID), zap.String("project", params.ProjectID), zap.String("cluster", params.ClusterID), zap.Bool("sharedTier", params.SharedTier), ) // Get the table name from the params (assuming it's stored in params.Type) tableName := params.TableName if tableName == "" { return nil, fmt.Errorf("table name is required") } var baseURL string if isSQLsDataSourceV2(params.DataSource) { baseURL = s.getSQLsV2Path(params.TenantID, params.ClusterID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) } else { metaInfo, err := s.mataCache.GetMetaInfo(ctx, params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.NextGen, params.Byoc) if err != nil { log.Error("GetMetaInfo failed", zap.Error(err)) return nil, err } if metaInfo == nil { return nil, fmt.Errorf("meta info not found for cluster %s", params.ClusterID) } baseURL = s.getSQLsBaseURL( params.TenantID, metaInfo.O11yID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen, ) } // Construct the full table path tablePath := buildSQLsTablePath(baseURL, tableName, params.DataSource) alias := tableAlias(params.ClusterID, tableName) // Create a query to get the schema using DESCRIBE on the delta table // We'll use a subquery with LIMIT 1 to avoid scanning the entire table query := fmt.Sprintf( `SELECT column_name, data_type FROM information_schema.columns WHERE table_name = '%s'`, alias) // Define a struct to hold the query results type columnInfo struct { ColumnName string `db:"column_name"` DataType string `db:"data_type"` } // Execute the query using SelectSlowlogInDL var columns []columnInfo err := s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &columns, alias, tablePath, query) if err != nil { log.Error("Failed to get table schema", zap.String("tablePath", tablePath), zap.String("query", query), zap.Error(err)) return nil, fmt.Errorf("failed to get table schema: %w", err) } // Convert column info to FieldInfo fieldInfos := make([]FieldInfo, 0, len(columns)) for _, col := range columns { fieldInfos = append(fieldInfos, FieldInfo{ Name: col.ColumnName, Type: col.DataType, }) } log.Info("ListFields finished", zap.String("table", tableName), zap.Int("fieldCount", len(fieldInfos))) return fieldInfos, nil } // ListTableDataResponse represents the response structure for ListTableData type ListTableDataResponse struct { Data []map[string]interface{} `json:"data"` Pagination struct { Total int64 `json:"total"` Limit int `json:"limit"` Offset int `json:"offset"` } `json:"pagination"` } func (s *DuckDBAdapter) ListTableData(ctx context.Context, params *ListTableDataParams) (*ListTableDataResponse, error) { log.Info("QueryDuckDBAdapter ListTableData called", zap.String("provider", params.Provider), zap.String("region", params.Region), zap.String("tenant", params.TenantID), zap.String("project", params.ProjectID), zap.String("cluster", params.ClusterID), zap.Bool("sharedTier", params.SharedTier), zap.Bool("nextGen", params.NextGen), zap.Bool("byoc", params.Byoc), ) // Get the table name from the params tableName := params.TableName if tableName == "" { return nil, fmt.Errorf("table name is required") } var baseURL string if isSQLsDataSourceV2(params.DataSource) { baseURL = s.getSQLsV2Path(params.TenantID, params.ClusterID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) } else { metaInfo, err := s.mataCache.GetMetaInfo(ctx, params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.NextGen, params.Byoc) if err != nil { log.Error("GetMetaInfo failed", zap.Error(err)) return nil, err } if metaInfo == nil { return nil, fmt.Errorf("meta info not found for cluster %s", params.ClusterID) } baseURL = s.getSQLsBaseURL( params.TenantID, metaInfo.O11yID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen, ) } // Construct the full table path tablePath := buildSQLsTablePath(baseURL, tableName, params.DataSource) // Build the SELECT fields. // params.Fields may be field names (mapped via sqlsFieldMap) or raw SQL expressions // already produced by convertStatementFields / filterStatementsFieldExprs. selectFields := "*" if len(params.Fields) > 0 { mappedFields := make([]string, 0, len(params.Fields)) fieldMap := sqlsFieldMap(params.DataSource) for _, field := range params.Fields { if isStatementsSQLExpression(field) { mappedFields = append(mappedFields, field) continue } if mapped, ok := fieldMap[field]; ok { mappedFields = append(mappedFields, mapped) } else { mappedFields = append(mappedFields, field) } } selectFields = strings.Join(mappedFields, ", ") } // Build the WHERE clause whereClause := "1=1" if params.StartTime != "" && params.EndTime != "" { whereClause += fmt.Sprintf(" AND time >= '%s' AND time <= '%s'", params.StartTime, params.EndTime) } if params.Filter != "" { whereClause += " AND " + params.Filter } // Build the ORDER BY clause orderByClause := "" if params.Sort != "" { orderByClause = " ORDER BY " + parseSortParameter(params.Sort) } // Build the GROUP BY clause groupByClause := "" if params.GroupBy != "" { groupByClause = " GROUP BY " + params.GroupBy } // Build the LIMIT clause limitClause := "" if params.Limit > 0 { limitClause = fmt.Sprintf(" LIMIT %d", params.Limit) if params.Offset > 0 { limitClause += fmt.Sprintf(" OFFSET %d", params.Offset) } } else { limitClause = fmt.Sprintf(" LIMIT %d", 100) } fromClause := fmt.Sprintf("delta_scan('%s')", tablePath) // Build the complete query query := fmt.Sprintf("SELECT %s FROM %s WHERE %s%s%s%s", selectFields, fromClause, whereClause, groupByClause, orderByClause, limitClause) log.Info("Executing query", zap.String("query", query)) // Execute the main query using the interface method var results []map[string]interface{} err := s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &results, "", tablePath, query) if err != nil { log.Error("Failed to execute data query", zap.String("tablePath", tablePath), zap.String("query", query), zap.Error(err)) return nil, fmt.Errorf("failed to execute data query: %w", err) } // Get total count countQuery := fmt.Sprintf("SELECT COUNT(*) as total FROM %s WHERE %s%s", fromClause, whereClause, groupByClause) type countResult struct { Total int64 `db:"total"` } var countResults []countResult err = s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &countResults, "", tablePath, countQuery) if err != nil { log.Error("Failed to get total count", zap.Error(err)) return nil, fmt.Errorf("failed to get total count: %w", err) } total := int64(0) if len(countResults) > 0 { total = countResults[0].Total } // Build response with pagination info response := &ListTableDataResponse{ Data: results, } response.Pagination.Total = total response.Pagination.Limit = params.Limit if params.Limit <= 0 { response.Pagination.Limit = 100 // default limit } response.Pagination.Offset = params.Offset log.Info("ListTableData finished", zap.String("table", tableName), zap.Int("resultCount", len(results)), zap.Int64("total", total), zap.Int("limit", response.Pagination.Limit), zap.Int("offset", response.Pagination.Offset)) // Return the new format with pagination info return response, nil } // GetTableDataByID retrieves a single record by ID func (s *DuckDBAdapter) GetTableDataByID(ctx context.Context, params *GetTableDataByIDParams) (map[string]interface{}, error) { // Get the table name from the params tableName := params.TableName if tableName == "" { return nil, fmt.Errorf("table name is required") } var baseURL string if isSQLsDataSourceV2(params.DataSource) { baseURL = s.getSQLsV2Path(params.TenantID, params.ClusterID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) } else { metaInfo, err := s.mataCache.GetMetaInfo(ctx, params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, params.NextGen, params.Byoc) if err != nil { log.Error("GetMetaInfo failed", zap.Error(err)) return nil, err } if metaInfo == nil { return nil, fmt.Errorf("meta info not found for cluster %s", params.ClusterID) } baseURL = s.getSQLsBaseURL( params.TenantID, metaInfo.O11yID, params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen, ) } // Construct the full table path tablePath := buildSQLsTablePath(baseURL, tableName, params.DataSource) alias := tableAlias(params.ClusterID, tableName) // Build the query to get a single record by ID query := fmt.Sprintf("SELECT * FROM %s WHERE id = %d LIMIT 1", alias, params.ID) log.Info("Executing GetTableDataByID query", zap.String("query", query)) // Execute the query using the interface method var results []map[string]interface{} err := s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &results, alias, tablePath, query) if err != nil { log.Error("Failed to execute GetTableDataByID query", zap.String("tablePath", tablePath), zap.String("query", query), zap.Error(err)) return nil, fmt.Errorf("failed to execute GetTableDataByID query: %w", err) } if len(results) == 0 { return nil, fmt.Errorf("record with ID %d not found", params.ID) } log.Info("GetTableDataByID finished", zap.String("table", tableName), zap.Int64("id", params.ID)) return results[0], nil } // parseSortParameter parses sort parameter like "Query_time:desc" or "Time:asc" // Examples: // // "Query_time:desc" -> "Query_time DESC" // "Time:asc" -> "Time ASC" // "Query_time" -> "Query_time ASC" (default to ASC) func parseSortParameter(sortParam string) string { if sortParam == "" { return "" } // Check if the parameter contains a colon if strings.Contains(sortParam, ":") { parts := strings.Split(sortParam, ":") if len(parts) == 2 { field := strings.TrimSpace(parts[0]) direction := strings.ToUpper(strings.TrimSpace(parts[1])) // Validate direction if direction == "ASC" || direction == "DESC" { return fmt.Sprintf("%s %s", field, direction) } } } // Default to ASC if no valid direction specified return fmt.Sprintf("%s ASC", sortParam) } func (s *DuckDBAdapter) StatementsAvailableFields(ctx context.Context) ([]string, error) { return nil, nil } func statementsSumRUExpr() string { return "sum(AVG_REQUEST_UNIT_READ*exec_count) + sum(AVG_REQUEST_UNIT_WRITE*exec_count)" } func statementsSumRUExprV2() string { return "sum(sum_rru) + sum(sum_wru)" } func statementsSumRUV2Expr() string { return "sum(COALESCE(AVG_REQUEST_UNIT_V2,0)*exec_count)" } func statementsSumRUV2ExprV2() string { return "sum(sum_ruv2)" } func statementsPickByDataSource(v1, v2, dataSource string) string { if isSQLsDataSourceV2(dataSource) { return v2 } return v1 } // staticOrderMetricExprs are order_by metrics with the same SQL for v1 and v2. var staticOrderMetricExprs = map[string]string{ "exec_count": "sum(exec_count)", "sum_latency": "sum(sum_latency)", "max_latency": "max(max_latency)", "min_latency": "min(min_latency)", "avg_latency": "sum(sum_latency)/sum(exec_count)", "sum_errors": "sum(sum_errors)", "sum_warnings": "sum(sum_warnings)", "plan_count": "count(DISTINCT plan_digest)", } // dualSourceOrderMetricExprs maps order_by to [v1 expr, v2 expr]. var dualSourceOrderMetricExprs = map[string][2]string{ "sum_ru": { statementsSumRUExpr(), statementsSumRUExprV2(), }, "sum_ru_v2": { statementsSumRUV2Expr(), statementsSumRUV2ExprV2(), }, "sum_rru": { "sum(AVG_REQUEST_UNIT_READ*exec_count)", "sum(sum_rru)", }, "sum_wru": { "sum(AVG_REQUEST_UNIT_WRITE*exec_count)", "sum(sum_wru)", }, "avg_ru": { "(sum(AVG_REQUEST_UNIT_READ*exec_count) + sum(AVG_REQUEST_UNIT_WRITE*exec_count))/sum(exec_count)", "(sum(sum_rru) + sum(sum_wru))/sum(exec_count)", }, "avg_ru_v2": { "sum(COALESCE(AVG_REQUEST_UNIT_V2,0)*exec_count)/sum(exec_count)", "sum(sum_ruv2)/sum(exec_count)", }, "avg_rru": { "sum(AVG_REQUEST_UNIT_READ*exec_count)/sum(exec_count)", "sum(sum_rru)/sum(exec_count)", }, "avg_wru": { "sum(AVG_REQUEST_UNIT_WRITE*exec_count)/sum(exec_count)", "sum(sum_wru)/sum(exec_count)", }, "max_ru": { "max(MAX_REQUEST_UNIT_READ + MAX_REQUEST_UNIT_WRITE)", "max(max_rru + max_wru)", }, "max_ru_v2": { "max(COALESCE(MAX_REQUEST_UNIT_V2,0))", "max(max_ruv2)", }, "max_rru": { "max(MAX_REQUEST_UNIT_READ)", "max(max_rru)", }, "max_wru": { "max(MAX_REQUEST_UNIT_WRITE)", "max(max_wru)", }, } func statementsOrderMetricExpr(orderBy, dataSource string) (string, error) { if expr, ok := staticOrderMetricExprs[orderBy]; ok { return expr, nil } if pair, ok := dualSourceOrderMetricExprs[orderBy]; ok { return statementsPickByDataSource(pair[0], pair[1], dataSource), nil } return "", fmt.Errorf("unsupported order_by: %s", orderBy) } func statementsSummaryStepInterval(step int) string { if step <= 0 { return "60 seconds" } return fmt.Sprintf("%d seconds", step) } func (s *DuckDBAdapter) resolvePersistedStatementsTable(ctx context.Context, p *StatementsSummaryParams) (alias, tablePath string, err error) { if isSQLsDataSourceV2(p.DataSource) { baseURL := s.getSQLsV2Path(p.TenantID, p.ClusterID, p.Env, p.Provider, p.Region, p.SharedTier, p.NextGen) alias = tableAlias(p.ClusterID, STATEMENTS_TABLE_NAME) return alias, buildSQLsTablePath(baseURL, STATEMENTS_TABLE_NAME, p.DataSource), nil } metaInfo, err := s.mataCache.GetMetaInfo(ctx, p.Env, p.Region, p.TenantID, p.ProjectID, p.ClusterID, p.NextGen, p.Byoc) if err != nil { return "", "", err } if metaInfo == nil { return "", "", fmt.Errorf("meta info not found for cluster %s", p.ClusterID) } baseURL := s.getSQLsBaseURL( p.TenantID, metaInfo.O11yID, p.Env, p.Provider, p.Region, p.SharedTier, p.NextGen, ) tablePath = buildSQLsTablePath(baseURL, STATEMENTS_TABLE_NAME, p.DataSource) alias = tableAlias(p.ClusterID, STATEMENTS_TABLE_NAME) return alias, tablePath, nil } type statementsSummaryBucketRow struct { EntityKey *string `db:"entity_key"` EntityLabel *string `db:"entity_label"` TimeWindow *time.Time `db:"time_window"` SeriesValue float64 `db:"series_value"` BucketRU float64 `db:"bucket_ru"` TotalExecCount *float64 `db:"total_exec_count"` TotalDuration *float64 `db:"total_duration_ns"` EntityOrderMetric *float64 `db:"entity_order_metric"` } // statementsTopStatsRow is scanned from DuckDB. Typed scanning avoids losing aggregate // values when drivers return non-primitive types into map[string]interface{}. type statementsTopStatsRow struct { Key *string `db:"key"` User *string `db:"user"` StmtType *string `db:"stmt_type"` DB *string `db:"db"` Table *string `db:"table"` SQLDigest *string `db:"sql_digest"` DigestText *string `db:"digest_text"` QuerySampleText *string `db:"query_sample_text"` PlanCount *int64 `db:"plan_count"` TotalRU float64 `db:"total_ru"` ExecCount *float64 `db:"exec_count"` SumLatency *float64 `db:"sum_latency"` OrderMetric float64 `db:"order_metric"` } func derefString(s *string) string { if s == nil { return "" } return *s } func float64PtrToInt64(v *float64) int64 { if v == nil { return 0 } return int64(*v) } func int64PtrVal(v *int64) int64 { if v == nil { return 0 } return *v } func statementsDigestTextExpr(dataSource string) string { if isSQLsDataSourceV2(dataSource) { return "normalized_sql" } return "digest_text" } func statementsQuerySampleTextExpr(dataSource string) string { if isSQLsDataSourceV2(dataSource) { return "sample_sql" } return "query_sample_text" } // statementsDBNameColumn is the persisted statements column for database (schema) name. func statementsDBNameColumn(dataSource string) string { if isSQLsDataSourceV2(dataSource) { return "schema_name" } return "SCHEMA_NAME" } // statementsTableNamesColumn is the persisted statements column for accessed table names. func statementsTableNamesColumn(dataSource string) string { if isSQLsDataSourceV2(dataSource) { return "table_names" } return "table_names" } func statementsSummaryDimension(p *StatementsSummaryParams) string { if strings.TrimSpace(p.Dimension) == "" { return "digest" } return p.Dimension } // statementsSummaryOthersEntityKey is the internal bucket key for entities outside top N. // Empty entity_key rows compete for top N; if they miss top N they are folded into this bucket. // API rows use is_other=true and dimension-specific label "others" (never empty string). const statementsSummaryOthersEntityKey = "__others__" // statementsV2FirstUserKeyExpr returns the primary user key for v2 auth_users attribution. // v2 rows may list multiple users; metrics and sample_user filters use only the first key // so totals stay consistent across digest/user dimensions (same as list "user" field). func statementsV2FirstUserKeyExpr(alias string) string { return fmt.Sprintf( "coalesce(list_element(coalesce(json_keys(%s.auth_users), []), 1), '')", alias, ) } func statementsV2FirstUserKeyColumn() string { return "coalesce(list_element(coalesce(json_keys(auth_users), []), 1), '')" } // statementsSummarySourceCTE builds the first CTE that assigns entity_key for the chosen dimension. func statementsSummarySourceCTE(alias, dimension, where, dataSource string) (cteSQL, cteName string) { switch dimension { case "user": if isSQLsDataSourceV2(dataSource) { firstUser := statementsV2FirstUserKeyExpr(alias) return fmt.Sprintf(`filtered AS ( SELECT %s.*, %s AS entity_key FROM %s WHERE %s )`, alias, firstUser, alias, where), "filtered" } userCol := statementsSampleUserColumn(dataSource) return fmt.Sprintf(`filtered AS ( SELECT %s.*, coalesce(%s.%s, '') AS entity_key FROM %s WHERE %s )`, alias, alias, userCol, alias, where), "filtered" case "sqltypes": return fmt.Sprintf(`filtered AS ( SELECT %s.*, coalesce(stmt_type, '') AS entity_key FROM %s WHERE %s )`, alias, alias, where), "filtered" case "db": dbCol := statementsDBNameColumn(dataSource) return fmt.Sprintf(`filtered AS ( SELECT %s.*, coalesce(%s.%s, '') AS entity_key FROM %s WHERE %s )`, alias, alias, dbCol, alias, where), "filtered" case "table": tableCol := statementsTableNamesColumn(dataSource) return fmt.Sprintf(`filtered AS ( SELECT %s.*, coalesce(%s.%s, '') AS entity_key FROM %s WHERE %s )`, alias, alias, tableCol, alias, where), "filtered" default: return fmt.Sprintf(`filtered AS ( SELECT %s.*, coalesce(%s.digest, '') AS entity_key FROM %s WHERE %s )`, alias, alias, alias, where), "filtered" } } func statementsSummaryBucketEntityLabelExpr(dimension, dataSource string) string { if dimension == "digest" { return fmt.Sprintf("any_value(f.%s)", statementsDigestTextExpr(dataSource)) } return "any_value(f.entity_key)" } func buildStatementsSummaryQuery(alias, where string, p *StatementsSummaryParams) (string, error) { dimension := statementsSummaryDimension(p) orderMetricExpr, err := statementsOrderMetricExpr(p.OrderBy, p.DataSource) if err != nil { return "", err } bucketRUExpr := statementsSumRUExpr() timeColumn := "f.SUMMARY_BEGIN_TIME" if isSQLsDataSourceV2(p.DataSource) { bucketRUExpr = statementsSumRUExprV2() timeColumn = "to_timestamp(f.\"begin\")" } step := statementsSummaryStepInterval(p.Step) sourceCTE, sourceName := statementsSummarySourceCTE(alias, dimension, where, p.DataSource) entityLabelExpr := statementsSummaryBucketEntityLabelExpr(dimension, p.DataSource) query := fmt.Sprintf(` WITH %s, top_entities AS ( SELECT entity_key, %s AS order_metric FROM %s GROUP BY entity_key ORDER BY order_metric DESC LIMIT %d ), tagged AS ( SELECT f.*, CASE WHEN t.entity_key IS NOT NULL THEN f.entity_key ELSE '%s' END AS bucket_key FROM %s f LEFT JOIN top_entities t ON f.entity_key = t.entity_key ), entity_metrics AS ( SELECT bucket_key AS entity_key, %s AS order_metric FROM tagged f GROUP BY bucket_key ), bucketed AS ( SELECT f.bucket_key AS entity_key, CASE WHEN f.bucket_key = '%s' THEN 'others' ELSE %s END AS entity_label, time_bucket('%s', %s) AS time_window, %s AS series_value, %s AS bucket_ru, sum(f.exec_count) AS total_exec_count, sum(f.sum_latency) AS total_duration_ns FROM tagged f GROUP BY f.bucket_key, time_window ) SELECT b.entity_key, b.entity_label, b.time_window, b.series_value, b.bucket_ru, b.total_exec_count, b.total_duration_ns, em.order_metric AS entity_order_metric FROM bucketed b INNER JOIN entity_metrics em ON b.entity_key = em.entity_key ORDER BY em.order_metric DESC, b.entity_key, b.time_window ASC`, sourceCTE, orderMetricExpr, sourceName, p.Top, statementsSummaryOthersEntityKey, sourceName, orderMetricExpr, statementsSummaryOthersEntityKey, entityLabelExpr, step, timeColumn, orderMetricExpr, bucketRUExpr) return query, nil } func populateStatementsSummaryItemFields(item *StatementsSummaryItem, dimension, entityKey, entityLabel string) { item.Dimension = dimension if entityKey == statementsSummaryOthersEntityKey { item.IsOther = true switch dimension { case "user": item.User = "others" case "sqltypes": item.StmtType = "others" case "db": item.DB = "others" case "table": item.Table = "others" default: item.SQLText = "others" } return } switch dimension { case "user": item.User = entityKey case "sqltypes": item.StmtType = entityKey case "db": item.DB = entityKey case "table": item.Table = entityKey default: item.SQLDigest = entityKey item.SQLText = entityLabel } } func statementsSummaryItemSortKey(item StatementsSummaryItem) string { switch item.Dimension { case "user": return item.User case "sqltypes": return item.StmtType case "db": return item.DB case "table": return item.Table default: return item.SQLDigest } } func statementsSummaryItemIsOthers(item StatementsSummaryItem) bool { return item.IsOther } func aggregateStatementsSummaryRows(rows []statementsSummaryBucketRow, p *StatementsSummaryParams) []StatementsSummaryItem { dimension := statementsSummaryDimension(p) timeRangeSecs := p.EndUnix - p.StartUnix type aggKey string agg := make(map[aggKey]*StatementsSummaryItem) for _, r := range rows { entityKey := derefString(r.EntityKey) entityLabel := derefString(r.EntityLabel) k := aggKey(entityKey) if _, ok := agg[k]; !ok { item := &StatementsSummaryItem{ PlanDigest: "", TimestampSec: []int64{}, ValueByTime: []float64{}, } populateStatementsSummaryItemFields(item, dimension, entityKey, entityLabel) agg[k] = item } item := agg[k] if r.EntityOrderMetric != nil { item.OrderMetric = *r.EntityOrderMetric } item.TotalRU += r.BucketRU if r.TotalExecCount != nil { item.ExecCount += int64(*r.TotalExecCount) } if r.TotalDuration != nil { item.DurationNs += int64(*r.TotalDuration) } if r.TimeWindow != nil { item.TimestampSec = append(item.TimestampSec, r.TimeWindow.Unix()) item.ValueByTime = append(item.ValueByTime, r.SeriesValue) } } out := make([]StatementsSummaryItem, 0, len(agg)) for _, item := range agg { if timeRangeSecs > 0 { item.ExecCountPerSec = float64(item.ExecCount) / float64(timeRangeSecs) } if item.ExecCount > 0 && item.DurationNs > 0 { item.DurationPerExecMs = float64(item.DurationNs) / float64(item.ExecCount) / 1e6 } formatStatementsSummaryItemForAPI(item, p.DataSource) out = append(out, *item) } sort.Slice(out, func(i, j int) bool { iIsOthers := statementsSummaryItemIsOthers(out[i]) jIsOthers := statementsSummaryItemIsOthers(out[j]) if iIsOthers != jIsOthers { return !iIsOthers } if out[i].OrderMetric != out[j].OrderMetric { return out[i].OrderMetric > out[j].OrderMetric } return statementsSummaryItemSortKey(out[i]) < statementsSummaryItemSortKey(out[j]) }) return out } // QueryStatementsSummary returns TopRU-like rows grouped by the requested dimension. func (s *DuckDBAdapter) QueryStatementsSummary(ctx context.Context, p *StatementsSummaryParams) ([]StatementsSummaryItem, error) { alias, tablePath, err := s.resolvePersistedStatementsTable(ctx, p) if err != nil { return nil, err } where := "1=1 AND " + p.Filter query, err := buildStatementsSummaryQuery(alias, where, p) if err != nil { return nil, err } var rows []statementsSummaryBucketRow if err := s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &rows, alias, tablePath, query); err != nil { return nil, fmt.Errorf("statements summary query: %w", err) } return aggregateStatementsSummaryRows(rows, p), nil } // statementsSampleUserColumn is the persisted statements column for auth user(s). func statementsSampleUserColumn(dataSource string) string { if isSQLsDataSourceV2(dataSource) { return "auth_users" } return "sample_user" } func statementsTopStatsFromClause(alias, dimension, dataSource string) string { return alias } // QueryStatementsTopStats runs grouped ranking query for /statements/topstats. func (s *DuckDBAdapter) QueryStatementsTopStats(ctx context.Context, p *StatementsTopStatsParams) ([]StatementsTopStatsItem, error) { alias, tablePath, err := s.resolvePersistedStatementsTable(ctx, &StatementsSummaryParams{ Env: p.Env, Provider: p.Provider, Region: p.Region, TenantID: p.TenantID, ProjectID: p.ProjectID, ClusterID: p.ClusterID, SharedTier: p.SharedTier, NextGen: p.NextGen, Byoc: p.Byoc, DataSource: p.DataSource, }) if err != nil { return nil, err } where := "1=1 AND " + p.Filter orderMetricExpr, err := statementsOrderMetricExpr(p.OrderBy, p.DataSource) if err != nil { return nil, err } ru := statementsSumRUExpr() if isSQLsDataSourceV2(p.DataSource) { ru = statementsSumRUExprV2() } if p.OrderBy == "sum_ru_v2" { if isSQLsDataSourceV2(p.DataSource) { ru = statementsSumRUV2ExprV2() } else { ru = statementsSumRUV2Expr() } } selectClause := "" groupBy := "" fromClause := statementsTopStatsFromClause(alias, p.Dimension, p.DataSource) digestTextExpr := statementsDigestTextExpr(p.DataSource) querySampleExpr := statementsQuerySampleTextExpr(p.DataSource) switch p.Dimension { case "user": if isSQLsDataSourceV2(p.DataSource) { firstUser := statementsV2FirstUserKeyExpr(alias) selectClause = strings.Join([]string{ fmt.Sprintf("%s AS key", firstUser), fmt.Sprintf("%s AS user", firstUser), }, ", ") groupBy = firstUser } else { userCol := statementsSampleUserColumn(p.DataSource) selectClause = strings.Join([]string{ fmt.Sprintf("coalesce(%s, '') AS key", userCol), fmt.Sprintf("coalesce(%s, '') AS user", userCol), }, ", ") groupBy = "1" } case "sqltypes": selectClause = strings.Join([]string{ "coalesce(stmt_type, '') AS key", "coalesce(stmt_type, '') AS stmt_type", }, ", ") groupBy = "1" case "db": dbCol := statementsDBNameColumn(p.DataSource) selectClause = strings.Join([]string{ fmt.Sprintf("coalesce(%s, '') AS key", dbCol), fmt.Sprintf("coalesce(%s, '') AS db", dbCol), }, ", ") groupBy = "1" case "table": tableCol := statementsTableNamesColumn(p.DataSource) selectClause = strings.Join([]string{ fmt.Sprintf("coalesce(%s, '') AS key", tableCol), fmt.Sprintf("coalesce(%s, '') AS table", tableCol), }, ", ") groupBy = "1" case "digest": selectClause = strings.Join([]string{ "digest AS key", "digest AS sql_digest", fmt.Sprintf("any_value(%s) AS digest_text", digestTextExpr), fmt.Sprintf("any_value(%s) AS query_sample_text", querySampleExpr), "count(DISTINCT plan_digest) AS plan_count", }, ", ") groupBy = "digest" default: return nil, fmt.Errorf("unsupported dimension: %s", p.Dimension) } query := fmt.Sprintf(` SELECT %s, %s AS total_ru, sum(exec_count) AS exec_count, sum(sum_latency) AS sum_latency, %s AS order_metric FROM %s WHERE %s GROUP BY %s ORDER BY order_metric DESC LIMIT %d`, selectClause, ru, orderMetricExpr, fromClause, where, groupBy, p.Top) var rows []statementsTopStatsRow if err := s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &rows, alias, tablePath, query); err != nil { return nil, fmt.Errorf("statements topstats query: %w", err) } out := make([]StatementsTopStatsItem, 0, len(rows)) for _, r := range rows { item := StatementsTopStatsItem{ Key: derefString(r.Key), Dimension: p.Dimension, OrderBy: p.OrderBy, Value: r.OrderMetric, ExecCount: float64PtrToInt64(r.ExecCount), SumLatency: float64PtrToInt64(r.SumLatency), } if p.OrderBy == "sum_ru" || p.OrderBy == "sum_ru_v2" { item.TotalRU = r.TotalRU } switch p.Dimension { case "user": item.User = derefString(r.User) case "sqltypes": item.StmtType = derefString(r.StmtType) case "db": item.DB = derefString(r.DB) case "table": item.Table = derefString(r.Table) case "digest": item.SQLDigest = derefString(r.SQLDigest) item.DigestText = derefString(r.DigestText) item.QuerySampleText = derefString(r.QuerySampleText) item.PlanCount = int64PtrVal(r.PlanCount) } formatStatementsTopStatsItemForAPI(&item, p.DataSource) out = append(out, item) } return out, nil } type statementsFieldValueRow struct { Value *string `db:"value"` } func buildStatementsFieldValuesQuery(alias, tablePath, where, field string, limit int, dataSource string) (string, error) { field = normalizeStatementsFilterField(field) kind, err := statementsFieldFilterKindFor(field) if err != nil { return "", err } scan := fmt.Sprintf("delta_scan('%s')", tablePath) limitClause := fmt.Sprintf(" LIMIT %d", limit) switch kind { case statementsFieldFilterUser: if isSQLsDataSourceV2(dataSource) { return fmt.Sprintf(` SELECT DISTINCT %s AS value FROM %s AS %s WHERE %s ORDER BY value ASC%s`, statementsV2FirstUserKeyExpr(alias), scan, alias, where, limitClause), nil } userCol := statementsSampleUserColumn(dataSource) return fmt.Sprintf(` SELECT DISTINCT coalesce(%s.%s, '') AS value FROM %s AS %s WHERE %s ORDER BY value ASC%s`, alias, userCol, scan, alias, where, limitClause), nil default: col, err := statementsScalarFilterColumn(field, dataSource) if err != nil { return "", err } return fmt.Sprintf(` SELECT DISTINCT coalesce(%s.%s, '') AS value FROM %s AS %s WHERE %s ORDER BY value ASC%s`, alias, col, scan, alias, where, limitClause), nil } } // QueryStatementsFieldValues returns distinct values for a field in the time window. func (s *DuckDBAdapter) QueryStatementsFieldValues(ctx context.Context, p *StatementsFieldValuesParams) ([]string, error) { alias, tablePath, err := s.resolvePersistedStatementsTable(ctx, &StatementsSummaryParams{ Env: p.Env, Provider: p.Provider, Region: p.Region, TenantID: p.TenantID, ProjectID: p.ProjectID, ClusterID: p.ClusterID, SharedTier: p.SharedTier, NextGen: p.NextGen, Byoc: p.Byoc, DataSource: p.DataSource, }) if err != nil { return nil, err } where := "1=1 AND " + p.Filter query, err := buildStatementsFieldValuesQuery(alias, tablePath, where, p.Field, p.Limit, p.DataSource) if err != nil { return nil, err } var rows []statementsFieldValueRow if err := s.duckdb.SelectSlowlogInDLWithoutFilterPushdown(ctx, &rows, alias, tablePath, query); err != nil { return nil, fmt.Errorf("statements field values query: %w", err) } out := make([]string, 0, len(rows)) for _, r := range rows { value := derefString(r.Value) if isSQLsDataSourceV2(p.DataSource) { value = statementsDisplayString(value) } out = append(out, value) } return out, nil }