package slowquery import ( "bytes" "context" "database/sql" "errors" "fmt" "io" "os" "strings" "testing" "time" "github.com/stretchr/testify/assert" o11yv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/slowquery/slowstats" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/storage" "github.com/tidbcloud/observability/o11ydiagnosis/pkg/utils/metacache" mockduckdb "github.com/tidbcloud/observability/test/mocko11ydiagnosis/pkg/utils/duckdb" mockmetacache "github.com/tidbcloud/observability/test/mocko11ydiagnosis/pkg/utils/metacache" "go.uber.org/mock/gomock" v1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "sigs.k8s.io/controller-runtime/pkg/client/fake" ) func selectSlowlogInDLDoAndReturn(results []*SlowlogInDL) func(context.Context, interface{}, string, string, string) error { return func(_ context.Context, dest interface{}, _, _, _ string) error { if destPtr, ok := dest.(*[]*SlowlogInDL); ok { *destPtr = results } return nil } } type MockStorage struct { savedData map[string][]byte } func (m *MockStorage) ListPrefixObject(ctx context.Context, bucket string, prefix string) ([]storage.ObjectMeta, error) { //TODO implement me panic("implement me") } func (m *MockStorage) ListObject(ctx context.Context, bucket string) ([]storage.ObjectMeta, error) { //TODO implement me panic("implement me") } func (m *MockStorage) ListObjectsPaged(ctx context.Context, bucket string, prefix string, pageSize int, pageToken string) (objects []storage.ObjectMeta, nextPageToken string, err error) { //TODO implement me panic("implement me") } func (m *MockStorage) Mkdir(ctx context.Context, bucket, key string, body io.Reader) error { //TODO implement me panic("implement me") } func NewMockStorage() *MockStorage { return &MockStorage{ savedData: make(map[string][]byte), } } // Implement storage.Storage interface func (m *MockStorage) SaveObject(ctx context.Context, bucket, key string, body io.Reader) error { data, err := io.ReadAll(body) if err != nil { return err } m.savedData[key] = data return nil } func (m *MockStorage) DeleteObject(ctx context.Context, bucket, key string) error { delete(m.savedData, key) return nil } func (m *MockStorage) DeleteObjects(ctx context.Context, bucket string, keys []string) error { for _, key := range keys { delete(m.savedData, key) } return nil } // Implement storage.Storage interface func (m *MockStorage) Object(ctx context.Context, bucket, key string) (*storage.Object, error) { data, exists := m.savedData[key] if !exists { return nil, os.ErrNotExist } return &storage.Object{ Reader: io.NopCloser(bytes.NewReader(data)), }, nil } // Implement storage.Storage interface func (m *MockStorage) ObjectExists(ctx context.Context, bucket, key string) (bool, error) { _, exists := m.savedData[key] return exists, nil } // Implement storage.Storage interface func (m *MockStorage) ObjectSize(ctx context.Context, bucket, key string) (int64, error) { data, exists := m.savedData[key] if !exists { return 0, os.ErrNotExist } return int64(len(data)), nil } // Implement storage.Storage interface func (m *MockStorage) Objects(ctx context.Context, bucket, prefix string) ([]storage.ObjectMeta, error) { var result []storage.ObjectMeta for key, data := range m.savedData { if prefix == "" || (len(key) >= len(prefix) && key[:len(prefix)] == prefix) { result = append(result, storage.ObjectMeta{ Key: key, Size: int64(len(data)), }) } } return result, nil } // Implement storage.Storage interface func (m *MockStorage) ObjectsCallback(ctx context.Context, bucket, prefix string, callback func(storage.ObjectMeta) bool) error { for key, data := range m.savedData { if prefix == "" || (len(key) >= len(prefix) && key[:len(prefix)] == prefix) { if !callback(storage.ObjectMeta{ Key: key, Size: int64(len(data)), }) { break } } } return nil } // Implement storage.Storage interface func (m *MockStorage) ObjectsCallbackWithStartKey(ctx context.Context, bucket, prefix, startKey string, callback func(storage.ObjectMeta) bool) error { skip := startKey != "" for key, data := range m.savedData { if prefix == "" || (len(key) >= len(prefix) && key[:len(prefix)] == prefix) { if skip { if key == startKey { skip = false } continue } if !callback(storage.ObjectMeta{ Key: key, Size: int64(len(data)), }) { break } } } return nil } // Implement storage.Storage interface func (m *MockStorage) ObjectsDelimiter(ctx context.Context, bucket, prefix string) ([]string, error) { prefixes := make(map[string]struct{}) for key := range m.savedData { if prefix == "" || (len(key) >= len(prefix) && key[:len(prefix)] == prefix) { // Find the next '/' as the delimiter nextSlash := -1 for i := len(prefix); i < len(key); i++ { if key[i] == '/' { nextSlash = i break } } if nextSlash != -1 { prefixes[key[:nextSlash+1]] = struct{}{} } } } result := make([]string, 0, len(prefixes)) for p := range prefixes { result = append(result, p) } return result, nil } // TestBuildListSQL tests the buildListSQL function. func TestBuildListSQL(t *testing.T) { slowlogTable := "s3://bucket/deltalake/tenant/cluster/slowlogs" clusterID := "test_cluster" alias := fmt.Sprintf("delta_scan('%s')", slowlogTable) tests := []struct { name string params *ListParams expectedSQL string expectedAlias string err error }{ { name: "Basic query with time range and default limit/order", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND (is_internal = false OR is_internal IS NULL) ORDER BY time ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "With Text filter", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, Text: "select * from", ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND (lower(query) LIKE '%%select * from%%' OR lower(digest) LIKE '%%select * from%%') AND (is_internal = false OR is_internal IS NULL) ORDER BY time ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "With Digest filter", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, Digest: "test_digest_123", ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND digest = 'test_digest_123' AND (is_internal = false OR is_internal IS NULL) ORDER BY time ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "With DB filter", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, DB: []string{"db1", "db2"}, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND db IN ('db1', 'db2') AND (is_internal = false OR is_internal IS NULL) ORDER BY time ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "With Plans filter", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, Plans: []string{"plan1", "plan2"}, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND plan_digest IN ('plan1', 'plan2') AND (is_internal = false OR is_internal IS NULL) ORDER BY time ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "Show internal queries", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, ShowInternal: true, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' ORDER BY time ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "Order by QueryTime DESC", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, OrderBy: "query_time", IsDesc: true, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND (is_internal = false OR is_internal IS NULL) ORDER BY query_time DESC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "Order by TotalKeys ASC", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, OrderBy: "total_keys", IsDesc: false, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND (is_internal = false OR is_internal IS NULL) ORDER BY CAST(total_keys AS UBIGINT) ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, { name: "Order by ProcessKeys DESC with custom limit", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, OrderBy: "process_keys", IsDesc: true, Limit: 50, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND (is_internal = false OR is_internal IS NULL) ORDER BY CAST(process_keys AS UBIGINT) DESC LIMIT 50;", alias), expectedAlias: "", err: nil, }, { name: "Multiple filters combined", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, Text: "heavy query", DB: []string{"prod_db"}, Plans: []string{"abc", "def"}, OrderBy: "memory_max", IsDesc: true, Limit: 10, ShowInternal: false, ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND (lower(query) LIKE '%%heavy query%%' OR lower(digest) LIKE '%%heavy query%%') AND db IN ('prod_db') AND plan_digest IN ('abc', 'def') AND (is_internal = false OR is_internal IS NULL) ORDER BY CAST(mem_max AS UBIGINT) DESC LIMIT 10;", alias), expectedAlias: "", err: nil, }, { name: "With fields parameter", params: &ListParams{ BeginTime: 1678838400, EndTime: 1678924799, Fields: "user,host,query_time", ClusterID: clusterID, }, expectedSQL: fmt.Sprintf("SELECT id, conn_id, digest, time, query_time, mem_max, user, host FROM %s WHERE time >= 1678838400.000000 AND time <= 1678924799.000000 AND date >= '20230315' AND date <= '20230315' AND (is_internal = false OR is_internal IS NULL) ORDER BY time ASC LIMIT 100;", alias), expectedAlias: "", err: nil, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { query, aliasResult, err := buildListSQL(tt.params, slowlogTable) if tt.err != nil { assert.Equal(t, tt.err, err) assert.Equal(t, tt.expectedSQL, query) assert.Equal(t, tt.expectedAlias, aliasResult) } else { assert.NoError(t, err) assert.Equal(t, tt.expectedSQL, query) assert.Equal(t, tt.expectedAlias, aliasResult) } }) } } // TestDuckDBAdapterDetail tests the Detail method of DuckDBAdapter. func TestNewSlowQueryDuckDBAdapter(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() scheme := runtime.NewScheme() _ = o11yv1.AddToScheme(scheme) mockKubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(&o11yv1.ClusterList{ Items: []o11yv1.Cluster{ { ObjectMeta: v1.ObjectMeta{ Name: "cluster-123", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenant1", labelApplicationID: "12345", }, }, }, }, }).Build() mockStorage := new(MockStorage) // Mock for duckdb.NewSlowQueryDuckDBFetcher // We can't directly mock a constructor, so we'll test the panic path and assume // that if no panic, the duckdb fetcher was initialized correctly. // For actual fetcher calls, we'll use a mock within the adapter methods. // Test case: Successful initialization // The actual NewSlowQueryDuckDBFetcher is called inside, so we cannot mock it directly. // We ensure it doesn't panic and the internal duckdb instance is not nil. adapter := NewSlowQueryDuckDBAdapter("local", "us-east-1", mockStorage, mockKubeClient) assert.NotNil(t, adapter) assert.NotNil(t, adapter.duckdb) // Checks if the internal duckdb fetcher was set } // TestDuckDBAdapterList tests the List method of DuckDBAdapter. func TestDuckDBAdapterList(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockFetcher := mockduckdb.NewMockQueryDuckDBFetcherInterface(ctrl) mockMetaCache := mockmetacache.NewMockMetaCache(ctrl) adapter := &DuckDBAdapter{ vendor: "aws", region: "us-east-1", duckdb: mockFetcher, mataCache: mockMetaCache, } // Mock GetMetaInfo mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), "cluster-test", false, false). Return(&metacache.MetaInfo{O11yID: "cluster-abc"}, nil).AnyTimes() // Test case 1: Successful retrieval t.Run("Successful List", func(t *testing.T) { params := &ListParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", BeginTime: 1678838400, EndTime: 1678838400, IsDesc: true, Limit: 10, } slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) expectedQuery, expectedAlias, _ := buildListSQL(params, slowlogTablePath) println("-------Expected Query:", expectedQuery) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), expectedAlias, slowlogTablePath, expectedQuery). DoAndReturn(selectSlowlogInDLDoAndReturn([]*SlowlogInDL{ { Time: 1678838400.0, Query: sql.NullString{String: "SELECT 1;", Valid: true}, Digest: "digest1", }, })).Times(1) slowLogs, err := adapter.List(context.TODO(), params) assert.NoError(t, err) assert.Len(t, slowLogs, 1) assert.Equal(t, "digest1", slowLogs[0].Digest) assert.Equal(t, "SELECT 1;", slowLogs[0].Query) }) // Test case 2: SelectSlowlogInDLWithoutFilterPushdown returns an error t.Run("SelectSlowlogInDL Error", func(t *testing.T) { params := &ListParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", BeginTime: 1678838400, EndTime: 1678838400, } slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) expectedQuery, expectedAlias, _ := buildListSQL(params, slowlogTablePath) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), expectedAlias, slowlogTablePath, expectedQuery). Return(errors.New("duckdb select error")).Times(1) slowLogs, err := adapter.List(context.TODO(), params) assert.Error(t, err) assert.Contains(t, err.Error(), "failed to select slowlog from DuckDB: duckdb select error") assert.Nil(t, slowLogs) }) // Test case 3: buildListSQL returns an error (e.g., unsupported order by - now handled by fieldAdapt) // This test case is largely invalid now since fieldAdapt will convert, not error. // Leaving it for demonstration, but it would need to be re-evaluated if a new error condition arises. t.Run("buildListSQL Error (unsupported order by - will not error with current impl)", func(t *testing.T) { params := &ListParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", BeginTime: 1678838400, EndTime: 1678838400, OrderBy: "invalid_field", } // Since fieldAdapt won't error, we expect a call to the fetcher. slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) expectedQuery, expectedAlias, _ := buildListSQL(params, slowlogTablePath) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), expectedAlias, slowlogTablePath, expectedQuery). DoAndReturn(selectSlowlogInDLDoAndReturn([]*SlowlogInDL{})).Times(1) slowLogs, err := adapter.List(context.TODO(), params) assert.NoError(t, err) // Expect no error from buildListSQL anymore assert.Empty(t, slowLogs) }) // Test case 4: Empty results t.Run("Empty Results", func(t *testing.T) { params := &ListParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", BeginTime: 1678838400, EndTime: 1678838400, } slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) expectedQuery, expectedAlias, _ := buildListSQL(params, slowlogTablePath) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), expectedAlias, slowlogTablePath, expectedQuery). DoAndReturn(selectSlowlogInDLDoAndReturn([]*SlowlogInDL{})).Times(1) slowLogs, err := adapter.List(context.TODO(), params) assert.NoError(t, err) assert.Len(t, slowLogs, 0) }) // Test case 5: GetMetaInfo returns an error t.Run("GetMetaInfo Error", func(t *testing.T) { params := &ListParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "error", BeginTime: 1678838400, EndTime: 1678838400, } mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), "error", false, false). Return(nil, errors.New("meta info error")).Times(1) slowLogs, err := adapter.List(context.TODO(), params) assert.Error(t, err) assert.Contains(t, err.Error(), "meta info error") assert.Nil(t, slowLogs) }) } // TestDuckDBAdapterDetail tests the Detail method of DuckDBAdapter. func TestDuckDBAdapterDetail(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockFetcher := mockduckdb.NewMockQueryDuckDBFetcherInterface(ctrl) mockMetaCache := mockmetacache.NewMockMetaCache(ctrl) adapter := &DuckDBAdapter{ vendor: "aws", region: "us-east-1", duckdb: mockFetcher, mataCache: mockMetaCache, } // Mock GetMetaInfo mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), "cluster-test", false, false). Return(&metacache.MetaInfo{O11yID: "cluster-abc"}, nil).AnyTimes() // Test case 1: Successful retrieval t.Run("Successful Detail", func(t *testing.T) { params := &DetailParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", Timestamp: 1678838400.0, ConnID: 12345, Digest: "detail_digest", } slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) expectedQuery, expectedAlias, _ := buildDetailSQL(params, slowlogTablePath, allSlowlogDetailStorageColumns()) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), gomock.Any(), slowlogTablePath, gomock.Any()). DoAndReturn(func(_ context.Context, dest interface{}, _ string, _ string, query string) error { if strings.Contains(query, "information_schema.columns") { if destPtr, ok := dest.(*[]columnInfo); ok { for name := range allSlowlogDetailStorageColumns() { *destPtr = append(*destPtr, columnInfo{ColumnName: name}) } } return nil } return selectSlowlogInDLDoAndReturn([]*SlowlogInDL{ { Time: float64(1678838400.0), Query: sql.NullString{String: "SELECT * FROM users WHERE id = 1;", Valid: true}, Digest: "detail_digest", ConnID: sql.NullString{String: "12345", Valid: true}, }, })(context.TODO(), dest, expectedAlias, slowlogTablePath, expectedQuery) }).Times(2) slowLog, err := adapter.Detail(context.TODO(), params) assert.NoError(t, err) assert.NotNil(t, slowLog) assert.Equal(t, "detail_digest", slowLog.Digest) assert.Equal(t, "SELECT * FROM users WHERE id = 1;", slowLog.Query) assert.Equal(t, uint64(12345), slowLog.ConnID) }) // Test case 2: SelectSlowlogInDLWithoutFilterPushdown returns an error t.Run("SelectSlowlogInDL Error", func(t *testing.T) { params := &DetailParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", Timestamp: 1678838400.0, ConnID: 12345, Digest: "detail_digest", } slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), gomock.Any(), slowlogTablePath, gomock.Any()). DoAndReturn(func(_ context.Context, dest interface{}, _ string, _ string, query string) error { if strings.Contains(query, "information_schema.columns") { if destPtr, ok := dest.(*[]columnInfo); ok { for name := range allSlowlogDetailStorageColumns() { *destPtr = append(*destPtr, columnInfo{ColumnName: name}) } } return nil } return errors.New("duckdb detail select error") }).Times(2) slowLog, err := adapter.Detail(context.TODO(), params) assert.Error(t, err) assert.Contains(t, err.Error(), "failed to select slowlog detail from DuckDB Delta Lake (joined): duckdb detail select error") assert.Nil(t, slowLog) }) // Test case 3: Slow log not found (empty results) t.Run("Slow Log Not Found", func(t *testing.T) { params := &DetailParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", Timestamp: 1678838400.0, ConnID: 12345, Digest: "non_existent_digest", } slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) expectedQuery, expectedAlias, _ := buildDetailSQL(params, slowlogTablePath, allSlowlogDetailStorageColumns()) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), gomock.Any(), slowlogTablePath, gomock.Any()). DoAndReturn(func(_ context.Context, dest interface{}, _ string, _ string, query string) error { if strings.Contains(query, "information_schema.columns") { if destPtr, ok := dest.(*[]columnInfo); ok { for name := range allSlowlogDetailStorageColumns() { *destPtr = append(*destPtr, columnInfo{ColumnName: name}) } } return nil } return selectSlowlogInDLDoAndReturn([]*SlowlogInDL{})(context.TODO(), dest, expectedAlias, slowlogTablePath, expectedQuery) }).Times(2) slowLog, err := adapter.Detail(context.TODO(), params) assert.Error(t, err) assert.EqualError(t, err, "slow log for detail not found") assert.Nil(t, slowLog) }) // Test case 4: GetMetaInfo returns an error t.Run("GetMetaInfo Error", func(t *testing.T) { params := &DetailParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "error", Timestamp: 1678838400.0, ConnID: 12345, Digest: "detail_digest", } mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), "error", false, false). Return(nil, errors.New("meta info error")).Times(1) slowLog, err := adapter.Detail(context.TODO(), params) assert.Error(t, err) assert.Contains(t, err.Error(), "meta info error") assert.Nil(t, slowLog) }) } // TestDuckDBAdapterDatabases tests the Databases method of DuckDBAdapter. func TestDuckDBAdapterDatabases(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockFetcher := mockduckdb.NewMockQueryDuckDBFetcherInterface(ctrl) mockMetaCache := mockmetacache.NewMockMetaCache(ctrl) adapter := &DuckDBAdapter{ vendor: "aws", region: "us-east-1", mataCache: mockMetaCache, duckdb: mockFetcher, } // Test case 1: Successful retrieval t.Run("Successful Databases", func(t *testing.T) { cluster := slowstats.ClusterInfo{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", ProjectID: "test_project", Env: "test", SharedTier: false, } beginHourTs := int64(1678838400) // 2023-03-15 00:00:00 hours := 1 // Mock GetMetaInfo mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), "test", "test_tenant", "cluster-test", false, false). Return(&metacache.MetaInfo{O11yID: "cluster-abc"}, nil).Times(1) // Expected query slowlogTablePath := adapter.getDuckDBSlowLogBaseURL("test_tenant", "cluster-abc", "aws", "us-east-1", "test", false, false) alias := tableAlias("cluster-test") expectedQuery := fmt.Sprintf(` SELECT DISTINCT db FROM %s WHERE time >= %.6f AND time <= %.6f AND date >= '%s' AND date <= '%s' AND db IS NOT NULL AND db != '' ORDER BY db `, alias, 1678838400.0, 1678842000.0, "20230315", "20230315") // Create a variable to hold the expected results expectedResults := []*SlowlogInDL{ { DB: sql.NullString{String: "db1", Valid: true}, }, { DB: sql.NullString{String: "db2", Valid: true}, }, } // Set up the mock to use Do to populate the destination mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown( gomock.Any(), gomock.Any(), // This will match the dest parameter alias, slowlogTablePath, strings.TrimSpace(expectedQuery), ). DoAndReturn(func(ctx context.Context, dest interface{}, _, _, _ string) error { // Type assert dest to *[]*SlowlogInDL and set its value if destPtr, ok := dest.(*[]*SlowlogInDL); ok { *destPtr = expectedResults } return nil }). Times(1) // Mock database results // mockFetcher.EXPECT(). // SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), alias, slowlogTablePath, strings.TrimSpace(expectedQuery)). // Return([]*SlowlogInDL{ // { // DB: sql.NullString{String: "db1", Valid: true}, // }, // { // DB: sql.NullString{String: "db2", Valid: true}, // }, // }, nil).Times(1) // Call the method databases, err := adapter.Databases(context.TODO(), cluster, beginHourTs, hours, false) // Assertions assert.NoError(t, err) assert.Len(t, databases, 2) assert.Equal(t, []string{"db1", "db2"}, databases) }) // Test case 2: Empty results t.Run("Empty Results", func(t *testing.T) { cluster := slowstats.ClusterInfo{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", ProjectID: "test_project", Env: "test", SharedTier: false, } beginHourTs := int64(1678838400) // 2023-03-15 00:00:00 hours := 1 // Mock GetMetaInfo mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), "test", "test_tenant", "cluster-test", false, false). Return(&metacache.MetaInfo{O11yID: "cluster-abc"}, nil).Times(1) // Expected query slowlogTablePath := adapter.getDuckDBSlowLogBaseURL("test_tenant", "cluster-abc", "aws", "us-east-1", "test", false, false) alias := tableAlias("cluster-test") expectedQuery := fmt.Sprintf(` SELECT DISTINCT db FROM %s WHERE time >= %.6f AND time <= %.6f AND date >= '%s' AND date <= '%s' AND db IS NOT NULL AND db != '' ORDER BY db `, alias, 1678838400.0, 1678842000.0, "20230315", "20230315") // Mock empty results mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), alias, slowlogTablePath, strings.TrimSpace(expectedQuery)). DoAndReturn(selectSlowlogInDLDoAndReturn([]*SlowlogInDL{})).Times(1) // Call the method databases, err := adapter.Databases(context.TODO(), cluster, beginHourTs, hours, false) // Assertions assert.NoError(t, err) assert.Empty(t, databases) }) // Test case 3: GetMetaInfo error t.Run("GetMetaInfo Error", func(t *testing.T) { cluster := slowstats.ClusterInfo{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "error-cluster", ProjectID: "test_project", Env: "test", SharedTier: false, } beginHourTs := int64(1678838400) hours := 1 // Mock GetMetaInfo error mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), "test", "test_tenant", "error-cluster", false, false). Return(nil, errors.New("meta info error")).Times(1) // Call the method databases, err := adapter.Databases(context.TODO(), cluster, beginHourTs, hours, false) // Assertions assert.Error(t, err) assert.Contains(t, err.Error(), "meta info error") assert.Nil(t, databases) }) // Test case 4: SelectSlowlogInDLWithoutFilterPushdown error t.Run("SelectSlowlogInDL Error", func(t *testing.T) { cluster := slowstats.ClusterInfo{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", ProjectID: "test_project", Env: "test", SharedTier: false, } beginHourTs := int64(1678838400) hours := 1 // Mock GetMetaInfo mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), gomock.Any(), gomock.Any(), "test", "test_tenant", "cluster-test", false, false). Return(&metacache.MetaInfo{O11yID: "cluster-abc"}, nil).Times(1) // Expected query slowlogTablePath := adapter.getDuckDBSlowLogBaseURL("test_tenant", "cluster-abc", "aws", "us-east-1", "test", false, false) alias := tableAlias("cluster-test") expectedQuery := fmt.Sprintf(` SELECT DISTINCT db FROM %s WHERE time >= %.6f AND time <= %.6f AND date >= '%s' AND date <= '%s' AND db IS NOT NULL AND db != '' ORDER BY db `, alias, 1678838400.0, 1678842000.0, "20230315", "20230315") // Mock database error mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), alias, slowlogTablePath, strings.TrimSpace(expectedQuery)). Return(errors.New("database error")).Times(1) // Call the method databases, err := adapter.Databases(context.TODO(), cluster, beginHourTs, hours, false) // Assertions assert.Error(t, err) assert.Contains(t, err.Error(), "failed to query databases from DuckDB: database error") assert.Nil(t, databases) }) } // TestGetDuckDBSlowLogURLs tests the URL construction function. func TestGetDuckDBSlowLogURLs(t *testing.T) { tests := []struct { name string env string provider string region string sharedTier bool tenantID string o11yID string expected string }{ { name: "AWS Prod Dedicated Tier", env: "prod", provider: "aws", region: "ap-east-1", sharedTier: false, tenantID: "t123", o11yID: "c456", expected: "s3://o11y-prod-shared-ap-east-1/deltalake/t123/c456/slowlogs", }, { name: "Azure Prod Shared Tier", env: "prod", provider: "azure", region: "eastus2", sharedTier: true, tenantID: "t345", o11yID: "c678", expected: "az://o11y-prod-shared-eastus2/deltalake/t345/c678/slowlogs", }, { name: "Unsupported Provider", env: "prod", provider: "unsupported", region: "us-east-1", sharedTier: false, tenantID: "t123", o11yID: "c456", expected: "", // Expected empty string for unsupported provider }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { adapter := &DuckDBAdapter{} // Only env needed for this function url := adapter.getDuckDBSlowLogBaseURL(tt.tenantID, tt.o11yID, tt.env, tt.provider, tt.region, tt.sharedTier, false) assert.Equal(t, tt.expected, url) }) } } // TestGetClusterTables tests the getClusterTables function. func TestGetClusterTables(t *testing.T) { scheme := runtime.NewScheme() _ = o11yv1.AddToScheme(scheme) tests := []struct { name string clusters []runtime.Object expectedTables map[string]string expectErr bool }{ { name: "Multiple valid clusters", clusters: []runtime.Object{ &o11yv1.Cluster{ ObjectMeta: v1.ObjectMeta{ Name: "cluster-001", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenantA", labelApplicationID: "111", }, }, }, &o11yv1.Cluster{ ObjectMeta: v1.ObjectMeta{ Name: "cluster-002", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenantB", labelApplicationID: "222", }, }, }, &o11yv1.Cluster{ ObjectMeta: v1.ObjectMeta{ Name: "cluster-003", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenantC", labelApplicationID: "333", }, }, }, }, expectedTables: map[string]string{ "slowlog_111": "s3://o11y-test-shared-us-east-1/deltalake/tenantA/cluster-001/slowlogs", "slowlog_222": "s3://o11y-test-shared-us-east-1/deltalake/tenantB/cluster-002/slowlogs", "slowlog_333": "s3://o11y-test-shared-us-east-1/deltalake/tenantC/cluster-003/slowlogs", }, expectErr: false, }, { name: "No valid clusters", clusters: []runtime.Object{ &o11yv1.Cluster{ // Incorrect bizType ObjectMeta: v1.ObjectMeta{ Name: "cluster-bad-biztype", Labels: map[string]string{ labelBizType: "other-biz", labelTenantID: "tenantX", labelApplicationID: "444", }, }, }, &o11yv1.Cluster{ // Invalid applicationID (not pure digits) ObjectMeta: v1.ObjectMeta{ Name: "cluster-bad-appid", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenantY", labelApplicationID: "abc", }, }, }, &o11yv1.Cluster{ // Excluded applicationID ObjectMeta: v1.ObjectMeta{ Name: "cluster-dataapp", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenantZ", labelApplicationID: "dataapp", }, }, }, }, expectedTables: map[string]string{}, expectErr: false, }, { name: "Empty cluster list", clusters: []runtime.Object{}, expectedTables: map[string]string{}, expectErr: false, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { mockKubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects(tt.clusters...).Build() adapter := &DuckDBAdapter{ vendor: "aws", region: "us-east-1", kubeCli: mockKubeClient, } tables, err := adapter.getClusterTables(context.Background()) if tt.expectErr { assert.Error(t, err) } else { assert.NoError(t, err) assert.Equal(t, tt.expectedTables, tables) } }) } } // TestMapSlowlogInDLToSlowLog tests the mapping function. func TestMapSlowlogInDLToSlowLog(t *testing.T) { now := time.Now().UnixMilli() / 1000.0 // Current Unix timestamp in float64 for testing dl := &SlowlogInDL{ Time: float64(now), Digest: "test_digest_xyz", Instance: sql.NullString{String: "instance-001", Valid: true}, TxnStartTS: sql.NullString{String: "123456789012345678", Valid: true}, User: sql.NullString{String: "root", Valid: true}, Host: sql.NullString{String: "127.0.0.1", Valid: true}, ConnID: sql.NullString{String: "98765", Valid: true}, SessionAlias: sql.NullString{String: "session_alias_1", Valid: true}, ExecRetryCount: sql.NullString{String: "5", Valid: true}, ExecRetryTime: sql.NullFloat64{Float64: 0.02, Valid: true}, QueryTime: sql.NullFloat64{Float64: 0.5, Valid: true}, ParseTime: sql.NullFloat64{Float64: 0.01, Valid: true}, CompileTime: sql.NullFloat64{Float64: 0.03, Valid: true}, RewriteTime: sql.NullFloat64{Float64: 0.005, Valid: true}, PreprocSubqueries: sql.NullString{String: "2", Valid: true}, PreprocSubqueriesTime: sql.NullFloat64{Float64: 0.002, Valid: true}, OptimizeTime: sql.NullFloat64{Float64: 0.04, Valid: true}, WaitTS: sql.NullFloat64{Float64: 0.06, Valid: true}, PrewriteTime: sql.NullFloat64{Float64: 0.07, Valid: true}, WaitPrewriteBinlogTime: sql.NullFloat64{Float64: 0.001, Valid: true}, CommitTime: sql.NullFloat64{Float64: 0.08, Valid: true}, GetCommitTSTime: sql.NullFloat64{Float64: 0.003, Valid: true}, CommitBackoffTime: sql.NullFloat64{Float64: 0.004, Valid: true}, BackoffTypes: sql.NullString{String: "lock", Valid: true}, ResolveLockTime: sql.NullFloat64{Float64: 0.009, Valid: true}, LocalLatchWaitTime: sql.NullFloat64{Float64: 0.001, Valid: true}, WriteKeys: sql.NullInt64{Int64: 10, Valid: true}, WriteSize: sql.NullInt64{Int64: 100, Valid: true}, PrewriteRegion: sql.NullInt64{Int64: 1, Valid: true}, TxnRetry: sql.NullInt64{Int64: 0, Valid: true}, CopTime: sql.NullFloat64{Float64: 0.1, Valid: true}, ProcessTime: sql.NullFloat64{Float64: 0.05, Valid: true}, WaitTime: sql.NullFloat64{Float64: 0.005, Valid: true}, BackoffTime: sql.NullFloat64{Float64: 0.002, Valid: true}, LockkeysTime: sql.NullFloat64{Float64: 0.01, Valid: true}, RequestCount: sql.NullString{String: "1", Valid: true}, TotalKeys: sql.NullString{String: "1000", Valid: true}, ProcessKeys: sql.NullString{String: "500", Valid: true}, RocksdbDeleteSkippedCount: sql.NullString{String: "0", Valid: true}, RocksdbKeySkippedCount: sql.NullString{String: "0", Valid: true}, RocksdbBlockCacheHitCount: sql.NullString{String: "5", Valid: true}, RocksdbBlockReadCount: sql.NullString{String: "2", Valid: true}, RocksdbBlockReadByte: sql.NullString{String: "200", Valid: true}, DB: sql.NullString{String: "test_db", Valid: true}, IndexNames: sql.NullString{String: "idx_test", Valid: true}, IsInternal: sql.NullBool{Bool: false, Valid: true}, Stats: sql.NullString{String: "stats_info", Valid: true}, CopProcAvg: sql.NullFloat64{Float64: 0.01, Valid: true}, CopProcP90: sql.NullFloat64{Float64: 0.02, Valid: true}, CopProcMax: sql.NullFloat64{Float64: 0.03, Valid: true}, CopProcAddr: sql.NullString{String: "192.168.1.1:20160", Valid: true}, CopWaitAvg: sql.NullFloat64{Float64: 0.001, Valid: true}, CopWaitP90: sql.NullFloat64{Float64: 0.002, Valid: true}, CopWaitMax: sql.NullFloat64{Float64: 0.003, Valid: true}, CopWaitAddr: sql.NullString{String: "192.168.1.2:20160", Valid: true}, MemMax: sql.NullInt64{Int64: 1024, Valid: true}, DiskMax: sql.NullInt64{Int64: 2048, Valid: true}, KvTotal: sql.NullFloat64{Float64: 0.15, Valid: true}, PdTotal: sql.NullFloat64{Float64: 0.005, Valid: true}, BackoffTotal: sql.NullFloat64{Float64: 0.003, Valid: true}, WriteSQLResponseTotal: sql.NullFloat64{Float64: 0.012, Valid: true}, ResultRows: sql.NullInt64{Int64: 10, Valid: true}, Warnings: sql.NullString{String: "warning_msg", Valid: true}, BackoffDetail: sql.NullString{String: "backoff_detail_msg", Valid: true}, Prepared: sql.NullBool{Bool: true, Valid: true}, Succ: sql.NullBool{Bool: true, Valid: true}, IsExplicitTxn: sql.NullBool{Bool: false, Valid: true}, IsWriteCacheTable: sql.NullBool{Bool: false, Valid: true}, PlanFromCache: sql.NullBool{Bool: false, Valid: true}, PlanFromBinding: sql.NullBool{Bool: false, Valid: true}, HasMoreResults: sql.NullBool{Bool: false, Valid: true}, ResourceGroup: sql.NullString{String: "rg1", Valid: true}, RequestUnitRead: sql.NullFloat64{Float64: 1.5, Valid: true}, RequestUnitWrite: sql.NullFloat64{Float64: 0.5, Valid: true}, RequestUnitV2: sql.NullFloat64{Float64: 12.5, Valid: true}, TimeQueuedByRC: sql.NullFloat64{Float64: 0.001, Valid: true}, PlanDigest: sql.NullString{String: "plan_digest_abc", Valid: true}, BinaryPlan: sql.NullString{String: "binary_plan_data", Valid: true}, ReadPoolTaskDetails: sql.NullString{String: sampleReadPoolTaskDetails, Valid: true}, PrevStmt: sql.NullString{String: "PREV STMT;", Valid: true}, Query: sql.NullString{String: "SELECT * FROM test.t;", Valid: true}, Plan: sql.NullString{String: "TableScan(t)", Valid: true}, ServerlessTenantID: sql.NullString{String: "st_123", Valid: true}, ServerlessProjectID: sql.NullString{String: "sp_456", Valid: true}, ServerlessClusterID: sql.NullString{String: "sc_789", Valid: true}, IsTiflash: sql.NullBool{Bool: false, Valid: true}, TidbCPUTime: sql.NullFloat64{Float64: 0.26, Valid: true}, TikvCPUTime: sql.NullFloat64{Float64: 2.760360638, Valid: true}, IARemoteReadSegmentCount: sql.NullString{String: "12", Valid: true}, IARemoteReadSegmentSize: sql.NullString{String: "4096", Valid: true}, IARemoteReadSegmentWaitTime: sql.NullFloat64{Float64: 0.015, Valid: true}, } sl := mapSlowlogInDLToSlowLog(dl) assert.Equal(t, timeFromFloat(float64(now)), sl.Time) assert.Equal(t, "test_digest_xyz", sl.Digest) assert.Equal(t, sampleReadPoolTaskDetails, sl.ReadPoolTaskDetails) assert.Equal(t, "instance-001", sl.Instance) assert.Equal(t, uint64(123456789012345678), sl.TxnStartTs) assert.Equal(t, "root", sl.User) assert.Equal(t, "127.0.0.1", sl.Host) assert.Equal(t, uint64(98765), sl.ConnID) assert.Equal(t, "session_alias_1", sl.SessionAlias) assert.Equal(t, uint64(5), sl.ExecRetryCount) assert.Equal(t, 0.02, sl.ExecRetryTime) assert.Equal(t, 0.5, sl.QueryTime) assert.Equal(t, 0.01, sl.ParseTime) assert.Equal(t, 0.03, sl.CompileTime) assert.Equal(t, 0.005, sl.RewriteTime) assert.Equal(t, uint64(2), sl.PreprocSubqueries) assert.Equal(t, 0.002, sl.PreprocSubqueriesTime) assert.Equal(t, 0.04, sl.OptimizeTime) assert.Equal(t, 0.06, sl.WaitTs) assert.Equal(t, 0.07, sl.PrewriteTime) assert.Equal(t, 0.001, sl.WaitPrewriteBinlogTime) assert.Equal(t, 0.08, sl.CommitTime) assert.Equal(t, 0.003, sl.GetCommitTsTime) assert.Equal(t, 0.004, sl.CommitBackoffTime) assert.Equal(t, "lock", sl.BackoffTypes) assert.Equal(t, 0.009, sl.ResolveLockTime) assert.Equal(t, 0.001, sl.LocalLatchWaitTime) assert.Equal(t, uint64(10), sl.WriteKeys) assert.Equal(t, uint64(100), sl.WriteSize) assert.Equal(t, uint64(1), sl.PrewriteRegion) assert.Equal(t, uint64(0), sl.TxnRetry) assert.Equal(t, 0.1, sl.CopTime) assert.Equal(t, 0.05, sl.ProcessTime) assert.Equal(t, 0.005, sl.WaitTime) assert.Equal(t, 0.002, sl.BackoffTime) assert.Equal(t, 0.01, sl.LockKeysTime) assert.Equal(t, uint64(1), sl.RequestCount) assert.Equal(t, uint64(1000), sl.TotalKeys) assert.Equal(t, uint64(500), sl.ProcessKeys) assert.Equal(t, uint64(0), sl.RocksdbDeleteSkippedCount) assert.Equal(t, uint64(0), sl.RocksdbKeySkippedCount) assert.Equal(t, uint64(5), sl.RocksdbBlockCacheHitCount) assert.Equal(t, uint64(2), sl.RocksdbBlockReadCount) assert.Equal(t, uint64(200), sl.RocksdbBlockReadByte) assert.Equal(t, "test_db", sl.DB) assert.Equal(t, "idx_test", sl.IndexNames) assert.False(t, sl.IsInternal) assert.Equal(t, "stats_info", sl.Stats) assert.Equal(t, 0.01, sl.CopProcAvg) assert.Equal(t, 0.02, sl.CopProcP90) assert.Equal(t, 0.03, sl.CopProcMax) assert.Equal(t, "192.168.1.1:20160", sl.CopProcAddr) assert.Equal(t, 0.001, sl.CopWaitAvg) assert.Equal(t, 0.002, sl.CopWaitP90) assert.Equal(t, 0.003, sl.CopWaitMax) assert.Equal(t, "192.168.1.2:20160", sl.CopWaitAddr) assert.Equal(t, uint64(1024), sl.MemMax) assert.Equal(t, uint64(2048), sl.DiskMax) assert.Equal(t, 0.15, sl.KVTotal) assert.Equal(t, 0.005, sl.PDTotal) assert.Equal(t, 0.003, sl.BackoffTotal) assert.Equal(t, 0.012, sl.WriteSQLResponseTotal) assert.Equal(t, uint64(10), sl.ResultRows) assert.Equal(t, "warning_msg", sl.Warnings) assert.Equal(t, "backoff_detail_msg", sl.BackoffDetail) assert.True(t, sl.Prepared) assert.True(t, sl.Succ) assert.False(t, sl.IsExplicitTxn) assert.False(t, sl.IsWriteCacheTable) assert.False(t, sl.PlanFromCache) assert.False(t, sl.PlanFromBinding) assert.False(t, sl.HasMoreResults) assert.Equal(t, "rg1", sl.ResourceGroup) assert.Equal(t, 1.5, sl.RequestUnitRead) assert.Equal(t, 0.5, sl.RequestUnitWrite) assert.Equal(t, 12.5, sl.RequestUnitV2) assert.Equal(t, 12.5, computeRUv2(sl)) assert.Equal(t, 0.001, sl.TimeQueuedByRc) assert.Equal(t, 0.26, sl.TidbCPUTime) assert.Equal(t, 2.760360638, sl.TikvCPUTime) assert.Equal(t, uint64(12), sl.IARemoteReadSegmentCount) assert.Equal(t, uint64(4096), sl.IARemoteReadSegmentSize) assert.Equal(t, 0.015, sl.IARemoteReadSegmentWaitTime) assert.Equal(t, "plan_digest_abc", sl.PlanDigest) assert.Equal(t, "binary_plan_data", sl.BinaryPlan) assert.Equal(t, "PREV STMT;", sl.PrevStmt) assert.Equal(t, "SELECT * FROM test.t;", sl.Query) assert.Equal(t, "TableScan(t)", sl.Plan) assert.Equal(t, "st_123", sl.ServerlessTenantID) assert.Equal(t, "sp_456", sl.ServerlessProjectID) assert.Equal(t, "sc_789", sl.ServerlessClusterID) assert.False(t, sl.IsTiflash) // Verify that a complete, sanitized Delta slow-log row reaches the NGM // detail response together with the observed read-pool payload. resp := slowLogToRespItem(sl) assert.Equal(t, "test_digest_xyz", resp.Digest) assert.Equal(t, "SELECT * FROM test.t;", resp.Query) assert.Equal(t, 0.5, resp.QueryTime) assert.Equal(t, "test_db", resp.DB) assert.Equal(t, sampleReadPoolTaskDetails, resp.ReadPoolTaskDetails) // Test with all null/empty values dlEmpty := &SlowlogInDL{} slEmpty := mapSlowlogInDLToSlowLog(dlEmpty) assert.Equal(t, time.Unix(0, 0), slEmpty.Time) assert.Equal(t, "", slEmpty.Digest) assert.Equal(t, uint64(0), slEmpty.TxnStartTs) assert.False(t, slEmpty.IsInternal) assert.Equal(t, uint64(0), slEmpty.MemMax) assert.Equal(t, uint64(0), slEmpty.TotalKeys) assert.Equal(t, uint64(0), slEmpty.ProcessKeys) assert.Equal(t, "", slEmpty.Query) assert.Equal(t, "", slEmpty.ServerlessTenantID) assert.Equal(t, "", slEmpty.ServerlessProjectID) assert.Equal(t, "", slEmpty.ServerlessClusterID) assert.Equal(t, "", slEmpty.Plan) assert.Equal(t, 0.0, slEmpty.LockKeysTime) assert.Equal(t, uint64(0), slEmpty.WriteKeys) assert.Equal(t, uint64(0), slEmpty.WriteSize) assert.Equal(t, uint64(0), slEmpty.PrewriteRegion) assert.Equal(t, uint64(0), slEmpty.TxnRetry) assert.Equal(t, uint64(0), slEmpty.RequestCount) assert.Equal(t, uint64(0), slEmpty.RocksdbDeleteSkippedCount) assert.Equal(t, uint64(0), slEmpty.RocksdbKeySkippedCount) assert.Equal(t, uint64(0), slEmpty.RocksdbBlockCacheHitCount) assert.Equal(t, uint64(0), slEmpty.RocksdbBlockReadCount) assert.Equal(t, uint64(0), slEmpty.RocksdbBlockReadByte) assert.Equal(t, uint64(0), slEmpty.ResultRows) assert.False(t, slEmpty.Prepared) assert.False(t, slEmpty.Succ) assert.False(t, slEmpty.IsExplicitTxn) assert.False(t, slEmpty.IsWriteCacheTable) assert.False(t, slEmpty.PlanFromCache) assert.False(t, slEmpty.PlanFromBinding) assert.False(t, slEmpty.HasMoreResults) assert.Equal(t, 0.0, slEmpty.RequestUnitRead) assert.Equal(t, 0.0, slEmpty.RequestUnitWrite) assert.Equal(t, 0.0, slEmpty.TimeQueuedByRc) assert.False(t, slEmpty.IsTiflash) } // TestParseStringToUint64 tests the helper function parseStringToUint64. func TestParseStringToUint64(t *testing.T) { assert.Equal(t, uint64(123), parseStringToUint64("123")) assert.Equal(t, uint64(0), parseStringToUint64("abc")) assert.Equal(t, uint64(0), parseStringToUint64("")) assert.Equal(t, uint64(18446744073709551615), parseStringToUint64("18446744073709551615")) assert.Equal(t, uint64(0), parseStringToUint64("-10")) // Should return 0 for negative assert.Equal(t, uint64(0), parseStringToUint64("999999999999999999999999999999999")) // Out of range } // TestParseInt64ToUint64 tests the helper function parseInt64ToUint64. func TestParseInt64ToUint64(t *testing.T) { assert.Equal(t, uint64(123), parseInt64ToUint64(123)) assert.Equal(t, uint64(0), parseInt64ToUint64(-1)) // Should return 0 for negative assert.Equal(t, uint64(0), parseInt64ToUint64(0)) assert.Equal(t, uint64(9223372036854775807), parseInt64ToUint64(9223372036854775807)) } // TestNotImplementedFunctions ensures that unimplemented methods panic. func TestNotImplementedFunctions(t *testing.T) { adapter := &DuckDBAdapter{} t.Run("Analyze panics", func(t *testing.T) { assert.PanicsWithValue(t, "Analyze not implemented for SlowQueryDuckDBAdapter", func() { adapter.Analyze(context.Background(), nil) }) }) t.Run("Download panics", func(t *testing.T) { assert.PanicsWithValue(t, "Download not implemented for SlowQueryDuckDBAdapter", func() { adapter.Download(context.Background(), nil) }) }) } func TestDuckDBAdapterAvailableFields(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() mockFetcher := mockduckdb.NewMockQueryDuckDBFetcherInterface(ctrl) mockMetaCache := mockmetacache.NewMockMetaCache(ctrl) adapter := &DuckDBAdapter{ vendor: "aws", region: "us-east-1", duckdb: mockFetcher, mataCache: mockMetaCache, } params := &AvailableFieldsParams{ TenantID: "test_tenant", Provider: "aws", Region: "us-east-1", ClusterID: "cluster-test", Env: "prod", } slowlogTablePath := adapter.getDuckDBSlowLogBaseURL(params.TenantID, "cluster-abc", params.Env, params.Provider, params.Region, params.SharedTier, params.NextGen) alias := tableAlias(params.ClusterID) mockMetaCache.EXPECT(). GetMetaInfo(gomock.Any(), params.Env, params.Region, params.TenantID, params.ProjectID, params.ClusterID, false, false). Return(&metacache.MetaInfo{O11yID: "cluster-abc"}, nil) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), alias, slowlogTablePath, gomock.Any()). DoAndReturn(func(ctx context.Context, dest interface{}, a, path, query string) error { cols := dest.(*[]columnInfo) *cols = []columnInfo{ {ColumnName: "digest"}, {ColumnName: "query"}, {ColumnName: "conn_id"}, {ColumnName: "time"}, {ColumnName: "query_time"}, {ColumnName: "mem_max"}, {ColumnName: "request_unit_read"}, {ColumnName: "request_unit_write"}, {ColumnName: "tidb_cpu_time"}, } return nil }) fields, err := adapter.AvailableFields(context.Background(), params) assert.NoError(t, err) assert.Contains(t, fields, "digest") assert.Contains(t, fields, "connection_id") assert.Contains(t, fields, "timestamp") assert.Contains(t, fields, "memory_max") assert.Contains(t, fields, "ru") assert.Contains(t, fields, "tidb_cpu_time") assert.NotContains(t, fields, "ru_v2") assert.NotContains(t, fields, "tikv_cpu_time") } // TestTableAlias tests the tableAlias function. func TestTableAlias(t *testing.T) { assert.Equal(t, "slowlog_mycluster", tableAlias("mycluster")) assert.Equal(t, "slowlog_anotherid", tableAlias("anotherid")) assert.Equal(t, "slowlog_12345", tableAlias("12345")) } // TestIsPureDigits tests the isPureDigits function. func TestIsPureDigits(t *testing.T) { tests := []struct { name string input string expected bool }{ {"Pure digits", "12345", true}, {"Empty string", "", false}, {"With letters", "123a45", false}, {"With symbols", "123-45", false}, {"Negative number", "-123", false}, {"Zero", "0", true}, {"Leading/trailing spaces", " 123 ", false}, // strings.TrimSpace would be needed if spaces were allowed } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { assert.Equal(t, tt.expected, isPureDigits(tt.input)) }) } } // TestConvertDeltalakeFields tests the convertDeltalakeFields function. func TestConvertDeltalakeFields(t *testing.T) { tests := []struct { name string input string expected string }{ { name: "Empty fields string", input: "", expected: "id, conn_id, digest, time, query_time, mem_max", }, { name: "Some custom fields", input: "user,host,db", expected: "id, conn_id, digest, time, query_time, mem_max, user, host, db", }, { name: "Read pool task details", input: "read_pool_task_details", expected: "id, conn_id, digest, time, query_time, mem_max, read_pool_task_details", }, { name: "Fields with mapping", input: "success,lock_keys_time,plan", expected: "id, conn_id, digest, time, query_time, mem_max, succ, lockkeys_time, plan", }, { name: "Fields including default ones", input: "query_time,digest,user", // query_time and digest are default expected: "id, conn_id, digest, time, query_time, mem_max, user", }, { name: "All default fields", input: "connection_id,digest,timestamp,query_time,memory_max,ru", expected: "id, conn_id, digest, time, query_time, mem_max, (COALESCE(request_unit_read, 0) + COALESCE(request_unit_write, 0)) AS ru", }, { name: "All default fields with ru_v2", input: "connection_id,digest,timestamp,query_time,memory_max,ru_v2", expected: "id, conn_id, digest, time, query_time, mem_max, (COALESCE(request_unit_read_v2, 0) + COALESCE(request_unit_write_v2, 0)) AS ru_v2", }, { name: "Mixed custom, mapped, and default fields", input: "user,success,timestamp,txn_start_ts", expected: "id, conn_id, digest, time, query_time, mem_max, user, succ, txn_start_ts", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { actual := convertDeltalakeFields(tt.input) assert.Equal(t, tt.expected, actual) }) } } // TestFieldAdapt tests the fieldAdapt function. func TestFieldAdapt(t *testing.T) { tests := []struct { name string input string expected string }{ {"Mapped field: success", "success", "succ"}, {"Mapped field: lock_keys_time", "lock_keys_time", "lockkeys_time"}, {"Mapped field: plan", "plan", "plan"}, {"Mapped field: plan_from_biniding", "plan_from_biniding", "plan_from_binding"}, {"Mapped field: prewrite_region", "prewrite_region", "prewrite_region"}, {"Mapped field: timestamp", "timestamp", "time"}, {"Mapped field: connection_id", "connection_id", "conn_id"}, {"Mapped field: memory_max", "memory_max", "mem_max"}, {"Mapped field: ru", "ru", "ru"}, {"Mapped field: ru_v2", "ru_v2", "ru_v2"}, {"Unmapped field", "unknown_field", "unknown_field"}, {"Empty string", "", ""}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { assert.Equal(t, tt.expected, fieldAdapt(tt.input)) }) } } // TestWarmup tests the warmup method. func TestWarmup(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() scheme := runtime.NewScheme() _ = o11yv1.AddToScheme(scheme) mockFetcher := mockduckdb.NewMockQueryDuckDBFetcherInterface(ctrl) mockKubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects( &o11yv1.Cluster{ ObjectMeta: v1.ObjectMeta{ Name: "cluster-warmup-1", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenant_warmup_1", labelApplicationID: "12345", }, }, }, &o11yv1.Cluster{ ObjectMeta: v1.ObjectMeta{ Name: "cluster-warmup-2", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenant_warmup_2", labelApplicationID: "67890", }, }, }, ).Build() adapter := &DuckDBAdapter{ vendor: "aws", region: "us-east-1", duckdb: mockFetcher, kubeCli: mockKubeClient, } today := time.Now().Format("20060102") expectedQuery1 := fmt.Sprintf(warmupQuery, "slowlog_12345", today) expectedQuery2 := fmt.Sprintf(warmupQuery, "slowlog_67890", today) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), "slowlog_12345", gomock.Any(), expectedQuery1). DoAndReturn(selectSlowlogInDLDoAndReturn([]*SlowlogInDL{})).Times(1) mockFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), "slowlog_67890", gomock.Any(), expectedQuery2). DoAndReturn(selectSlowlogInDLDoAndReturn([]*SlowlogInDL{})).Times(1) err := adapter.warmup(context.Background(), mockFetcher) assert.NoError(t, err) } // TestRenewAndWarmupDuckDB tests the RenewAndWarmupDuckDB method. func TestRenewAndWarmupDuckDB(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() // Mock for the old and new fetchers mockOldFetcher := mockduckdb.NewMockQueryDuckDBFetcherInterface(ctrl) mockNewFetcher := mockduckdb.NewMockQueryDuckDBFetcherInterface(ctrl) // Initialize a scheme and add the o11yv1.Cluster type to it scheme := runtime.NewScheme() _ = o11yv1.AddToScheme(scheme) // This is crucial for fixing the error // Mock KubeClient to allow getClusterTables to succeed in warmup mockKubeClient := fake.NewClientBuilder().WithScheme(scheme).WithRuntimeObjects( &o11yv1.Cluster{ ObjectMeta: v1.ObjectMeta{ Name: "cluster-renew-1", Labels: map[string]string{ labelBizType: labelBizTypeValue, labelTenantID: "tenant_renew_1", labelApplicationID: "11111", }, }, }, ).Build() adapter := &DuckDBAdapter{ vendor: "aws", region: "us-east-1", duckdb: mockOldFetcher, // The adapter starts with an old mock fetcher kubeCli: mockKubeClient, } // Expect the old fetcher to be closed after a delay mockOldFetcher.EXPECT().Close(gomock.Any()).Return(nil).AnyTimes() // Use AnyTimes for background goroutine // Mock the call to SelectSlowlogInDLWithoutFilterPushdown during warmup for the new fetcher today := time.Now().Format("20060102") warmupQ := fmt.Sprintf(warmupQuery, "slowlog_11111", today) mockNewFetcher.EXPECT(). SelectSlowlogInDLWithoutFilterPushdown(gomock.Any(), gomock.Any(), "slowlog_11111", gomock.Any(), warmupQ). DoAndReturn(selectSlowlogInDLDoAndReturn([]*SlowlogInDL{})).AnyTimes() // AnyTimes as it's called internally by warmup err := adapter.RenewAndWarmupDuckDB(context.Background()) assert.NoError(t, err) // Simulate the delay for the old fetcher to be closed time.Sleep(time.Second * 6) // Wait slightly more than 5 minutes for the goroutine to run }