From 8fb89109b36e6b2d60cb2f97f1be002b136f3931 Mon Sep 17 00:00:00 2001 From: Ashish Kolhe <35160958+oasisk@users.noreply.github.com> Date: Mon, 5 Jan 2026 15:37:23 +0530 Subject: [PATCH] fix: 9658 - update query on stream change (#9823) ### **User description** fixes: https://github.com/openobserve/openobserve/issues/9658 ___ ### **PR Type** Bug fix, Tests ___ ### **Description** - Auto-update SQL query FROM clause on stream change - regenerate select * query - Add comprehensive Playwright E2E tests ___

File Walkthrough
Relevant files
Enhancement
ScheduledPipeline.vue
Add auto-update SQL FROM clause watcher                                   
web/src/components/pipeline/NodeForm/ScheduledPipeline.vue
  • Added watcher on selectedStreamName for SQL tab
  • Parsed and replaced FROM clause when stream changes
  • Ensured conditions: non-empty query and matching old stream
  • Graceful error handling on parse failures
 +57/-1   
Tests
scheduled-pipeline-stream-auto-update.spec.js
Add scheduled pipeline stream auto-update tests                   
tests/ui-testing/playwright-tests/Pipelines/scheduled-pipeline-stream-auto-update.spec.js
  • Introduce E2E tests verifying SQL auto-update behavior
  • Validate preservation of WHERE and complex clauses
  • Test manual edit protection and initial load cases
  • Skip PromQL tab scenario
 +480/-0 
___ --------- Co-authored-by: greptile-apps[bot] <165735046+greptile-apps[bot]@users.noreply.github.com> --- .../pages/pipelinesPages/pipelinesPage.js | 133 +++++++++++ ...eduled-pipeline-stream-auto-update.spec.js | 206 ++++++++++++++++++ .../pipeline/NodeForm/ScheduledPipeline.vue | 42 +++- 3 files changed, 373 insertions(+), 8 deletions(-) create mode 100644 tests/ui-testing/playwright-tests/Pipelines/scheduled-pipeline-stream-auto-update.spec.js diff --git a/tests/ui-testing/pages/pipelinesPages/pipelinesPage.js b/tests/ui-testing/pages/pipelinesPages/pipelinesPage.js index 68395b4f5f..015e75655e 100644 --- a/tests/ui-testing/pages/pipelinesPages/pipelinesPage.js +++ b/tests/ui-testing/pages/pipelinesPages/pipelinesPage.js @@ -98,6 +98,14 @@ export class PipelinesPage { this.submitButton = page.locator('[data-test="add-destination-submit-btn"]'); this.streamsMenuItem = page.locator('[data-test="menu-link-\\/streams-item"]'); this.refreshStatsButton = page.locator('[data-test="log-stream-refresh-stats-btn"]'); + + // Scheduled Pipeline Dialog selectors + this.scheduledPipelineTabs = page.locator('[data-test="scheduled-pipeline-tabs"]'); + this.scheduledPipelineSqlEditor = page.locator('[data-test="scheduled-pipeline-sql-editor"]'); + this.buildQuerySection = page.getByText('Build Query').first(); + this.streamTypeLabel = page.getByLabel(/Stream Type/i); + this.streamNameLabel = page.getByLabel(/Stream Name/i); + this.monacoEditorViewLines = page.locator('.monaco-editor .view-lines'); this.searchStreamInput = page.getByPlaceholder('Search Stream'); this.exploreButton = page.getByRole('button', { name: 'Explore' }); this.timestampColumnMenu = page.locator('[data-test="log-table-column-1-_timestamp"] [data-test="table-row-expand-menu"]'); @@ -1578,4 +1586,129 @@ export class PipelinesPage { } await this.saveQuery(); } + + // ======================================== + // Scheduled Pipeline Dialog Methods (POM Fix) + // ======================================== + + /** + * Wait for scheduled pipeline dialog to be visible + * Replaces: await page.locator('[data-test="scheduled-pipeline-tabs"]').waitFor() + */ + async waitForScheduledPipelineDialog() { + await this.scheduledPipelineTabs.waitFor({ state: 'visible', timeout: 10000 }); + testLogger.info('Scheduled pipeline dialog is visible'); + } + + /** + * Wait for SQL editor to be visible + * Replaces: await expect(page.locator('[data-test="scheduled-pipeline-sql-editor"]')).toBeVisible() + */ + async expectSqlEditorVisible() { + await expect(this.scheduledPipelineSqlEditor).toBeVisible({ timeout: 10000 }); + testLogger.info('SQL editor is visible'); + } + + /** + * Expand Build Query section + * Replaces: await page.getByText('Build Query').first().click() + */ + async expandBuildQuerySection() { + await this.buildQuerySection.waitFor({ state: 'visible', timeout: 5000 }); + await this.buildQuerySection.click(); + testLogger.info('Build Query section expanded'); + } + + /** + * Select stream type from dropdown + * @param {string} type - Stream type (e.g., 'logs') + * Replaces: await page.getByLabel(/Stream Type/i).click() and option selection + */ + async selectStreamType(type) { + testLogger.info(`Selecting stream type: ${type}`); + await this.streamTypeLabel.click(); + // Wait for dropdown to open + await this.page.waitForFunction(() => { + const options = document.querySelectorAll('[role="option"]'); + return options.length > 0; + }, { timeout: 3000 }); + await this.page.getByRole("option", { name: type, exact: true }).click(); + testLogger.info(`Stream type '${type}' selected`); + } + + /** + * Select stream name from dropdown + * @param {string} streamName - Stream name to select + * Replaces: await page.getByLabel(/Stream Name/i).click(), fill, and option selection + */ + async selectStreamName(streamName) { + testLogger.info(`Selecting stream: ${streamName}`); + await this.streamNameLabel.click(); + // Wait briefly for dropdown to open + await this.page.waitForTimeout(500); + await this.streamNameLabel.fill(streamName); + // Wait for options to filter + await this.page.waitForTimeout(1000); + await this.page.getByRole("option", { name: streamName, exact: true }).click(); + testLogger.info(`Stream '${streamName}' selected`); + } + + /** + * Get current query text from Monaco editor + * Replaces: await page.locator('.monaco-editor .view-lines').textContent() + * @returns {Promise} The query text + */ + async getQueryText() { + const text = await this.monacoEditorViewLines.textContent(); + testLogger.info(`Query text retrieved: ${text?.substring(0, 50)}...`); + return text; + } + + /** + * Wait for watcher to process stream change + * Deterministic wait that checks for query state to stabilize + * Replaces: await page.waitForTimeout(2000) after stream change + */ + async waitForStreamChangeWatcher() { + testLogger.info('Waiting for stream change watcher to process...'); + // Wait for Vue watcher to execute and update query + await this.page.waitForFunction(() => { + const editor = document.querySelector('.monaco-editor'); + return editor !== null; + }, { timeout: 3000 }); + // Additional small wait for query update to complete + await this.page.waitForTimeout(500); + testLogger.info('Watcher processing complete'); + } + + /** + * Expect query editor to contain specific text + * @param {string} expectedText - Text that should be in the query + */ + async expectQueryToContain(expectedText) { + const queryText = await this.getQueryText(); + expect(queryText).toContain(expectedText); + testLogger.info(`Query contains expected text: ${expectedText}`); + } + + /** + * Expect query editor to NOT contain specific text + * @param {string} unexpectedText - Text that should NOT be in the query + */ + async expectQueryNotToContain(unexpectedText) { + const queryText = await this.getQueryText(); + expect(queryText).not.toContain(unexpectedText); + testLogger.info(`Query does not contain: ${unexpectedText}`); + } + + /** + * Expect query to be empty or very short (cleared state) + * Replaces: expect(queryText.length).toBeLessThanOrEqual(10) + */ + async expectQueryCleared() { + const queryText = await this.getQueryText(); + const trimmed = queryText?.trim() || ''; + expect(trimmed.length).toBeLessThanOrEqual(10); + testLogger.info(`Query is cleared (length: ${trimmed.length})`); + } } \ No newline at end of file diff --git a/tests/ui-testing/playwright-tests/Pipelines/scheduled-pipeline-stream-auto-update.spec.js b/tests/ui-testing/playwright-tests/Pipelines/scheduled-pipeline-stream-auto-update.spec.js new file mode 100644 index 0000000000..26d25a410d --- /dev/null +++ b/tests/ui-testing/playwright-tests/Pipelines/scheduled-pipeline-stream-auto-update.spec.js @@ -0,0 +1,206 @@ +import { test, expect } from "../baseFixtures.js"; +import logData from "../../fixtures/log.json"; +import logsdata from "../../../test-data/logs_data.json"; +import PageManager from "../../pages/page-manager.js"; +import { LoginPage } from "../../pages/generalPages/loginPage.js"; +const testLogger = require('../utils/test-logger.js'); + +test.describe.configure({ mode: "serial" }); + +test.use({ + contextOptions: { + slowMo: 1000 + } +}); + +test.describe("Scheduled Pipeline Stream Auto-Update", { tag: ['@all', '@scheduledPipeline'] }, () => { + let pageManager; + let loginPage; + + test.beforeEach(async ({ page }, testInfo) => { + testLogger.testStart(testInfo.title, testInfo.file); + + // Login + loginPage = new LoginPage(page); + await loginPage.gotoLoginPage(); + await loginPage.loginAsInternalUser(); + await loginPage.login(); + + pageManager = new PageManager(page); + + // Ingest test data to ensure streams exist + const streamNames = ["e2e_automate", "k8s_json"]; + await pageManager.pipelinesPage.bulkIngestToStreams(streamNames, logsdata); + + await page.goto( + `${logData.logsUrl}?org_identifier=${process.env["ORGNAME"]}` + ); + + testLogger.info('Test setup completed'); + }); + + test.afterEach(async ({ page }, testInfo) => { + testLogger.testEnd(testInfo.title, testInfo.status); + }); + + // P0 - Smoke Tests + + test("should auto-generate SELECT * query when stream changes", { + tag: ['@smoke', '@P0'] + }, async ({ page }) => { + testLogger.info('Testing query is auto-generated when stream changes'); + + // Navigate to pipelines + await pageManager.pipelinesPage.openPipelineMenu(); + await page.waitForTimeout(1000); + + // Add new pipeline + await pageManager.pipelinesPage.addPipeline(); + + // Drag Query button to canvas - opens Query form dialog + await pageManager.pipelinesPage.dragStreamToTarget(pageManager.pipelinesPage.queryButton); + await page.waitForTimeout(500); + + // Wait for scheduled pipeline form dialog to load + await pageManager.pipelinesPage.waitForScheduledPipelineDialog(); + await page.waitForTimeout(1000); + + // Expand "Build Query" section + await pageManager.pipelinesPage.expandBuildQuerySection(); + await page.waitForTimeout(500); + + // Select stream type (logs) + await pageManager.pipelinesPage.selectStreamType('logs'); + await page.waitForTimeout(1000); + + // Select first stream (e2e_automate) + await pageManager.pipelinesPage.selectStreamName('e2e_automate'); + await page.waitForTimeout(1000); + + // Verify default query generated + await pageManager.pipelinesPage.expectSqlEditorVisible(); + + // Get query text from Monaco editor + await pageManager.pipelinesPage.expectQueryToContain('e2e_automate'); + + // Change stream to k8s_json + testLogger.info('Changing stream to k8s_json'); + await pageManager.pipelinesPage.selectStreamName('k8s_json'); + await pageManager.pipelinesPage.waitForStreamChangeWatcher(); + + // Verify query is auto-generated with new stream name + await pageManager.pipelinesPage.expectQueryToContain('SELECT'); + await pageManager.pipelinesPage.expectQueryToContain('FROM'); + await pageManager.pipelinesPage.expectQueryToContain('k8s_json'); + await pageManager.pipelinesPage.expectQueryNotToContain('e2e_automate'); + + testLogger.info('✅ Test passed: Query auto-generated correctly on stream change'); + }); + + // Note: WHERE clause preservation tests removed + // New behavior auto-generates fresh SELECT * query on stream change + // This prevents field reference issues when switching between streams with different schemas + + test("should handle initial stream selection (existing behavior)", { + tag: ['@regression', '@P1'] + }, async ({ page }) => { + testLogger.info('Testing initial stream selection generates default query'); + + await pageManager.pipelinesPage.openPipelineMenu(); + await page.waitForTimeout(1000); + await pageManager.pipelinesPage.addPipeline(); + + // Drag Query button to canvas - opens Query form dialog + await pageManager.pipelinesPage.dragStreamToTarget(pageManager.pipelinesPage.queryButton); + await page.waitForTimeout(500); + + // Wait for scheduled pipeline form dialog to load + await pageManager.pipelinesPage.waitForScheduledPipelineDialog(); + await page.waitForTimeout(1000); + + // Verify query is empty initially (or not visible) + testLogger.info('Verifying empty state'); + + // Expand "Build Query" section + await pageManager.pipelinesPage.expandBuildQuerySection(); + await page.waitForTimeout(500); + + // Select stream type (logs) + await pageManager.pipelinesPage.selectStreamType('logs'); + await page.waitForTimeout(1000); + + // Select first stream + await pageManager.pipelinesPage.selectStreamName('e2e_automate'); + await page.waitForTimeout(1000); + + // Verify default query generated + await pageManager.pipelinesPage.expectSqlEditorVisible(); + await pageManager.pipelinesPage.expectQueryToContain('SELECT'); + await pageManager.pipelinesPage.expectQueryToContain('FROM'); + await pageManager.pipelinesPage.expectQueryToContain('e2e_automate'); + + testLogger.info('✅ Test passed: Default query generated correctly'); + }); + + test.skip("should not update PromQL tab when stream changes", { + tag: ['@functional', '@P1'] + }, async ({ page }) => { + testLogger.info('Testing PromQL tab is unaffected by SQL watcher'); + + await pageManager.pipelinesPage.openPipelineMenu(); + await page.waitForTimeout(1000); + await pageManager.pipelinesPage.addPipeline(); + + // Drag Query button to canvas - opens Query form dialog + await pageManager.pipelinesPage.dragStreamToTarget(pageManager.pipelinesPage.queryButton); + await page.waitForTimeout(500); + + // Wait for scheduled pipeline form dialog to load + await pageManager.pipelinesPage.waitForScheduledPipelineDialog(); + await page.waitForTimeout(1000); + + // Expand "Build Query" section + await pageManager.pipelinesPage.expandBuildQuerySection(); + await page.waitForTimeout(500); + + // Select stream type (metrics) - PromQL only enabled for metrics + await pageManager.pipelinesPage.selectStreamType('metrics'); + await page.waitForTimeout(1000); + + // Select a metrics stream in SQL tab + testLogger.info('Selecting metrics stream'); + await pageManager.pipelinesPage.streamNameLabel.click(); + await page.waitForTimeout(500); + // Note: We'll use any available metrics stream + const streamOptions = page.getByRole("option").filter({ hasText: /^[a-z0-9_]+$/ }).first(); + await streamOptions.click(); + await page.waitForTimeout(2000); + + // Get SQL query text + await pageManager.pipelinesPage.expectQueryToContain('FROM'); + + // Switch to PromQL tab (now enabled because we selected metrics) + testLogger.info('Switching to PromQL tab'); + const promqlTab = pageManager.pipelinesPage.scheduledPipelineTabs.getByRole('button', { name: /PromQL/i }); + await promqlTab.waitFor({ state: 'visible', timeout: 5000 }); + await promqlTab.click(); + await page.waitForTimeout(1000); + + // Verify we're on PromQL tab + testLogger.info('Verifying PromQL tab active'); + + // Switch back to SQL tab + testLogger.info('Switching back to SQL tab'); + const sqlTab = pageManager.pipelinesPage.scheduledPipelineTabs.getByRole('button', { name: /SQL/i }); + await sqlTab.click(); + await page.waitForTimeout(1000); + + // Verify SQL query still exists (tab switching doesn't affect SQL query) + await pageManager.pipelinesPage.expectQueryToContain('FROM'); + + testLogger.info('✅ Test passed: PromQL tab independent of SQL watcher'); + }); + + // Note: Manual edit protection test removed + // New behavior always auto-generates SELECT * query on stream change for data safety +}); diff --git a/web/src/components/pipeline/NodeForm/ScheduledPipeline.vue b/web/src/components/pipeline/NodeForm/ScheduledPipeline.vue index 98c7c6b642..8dc9d60ed8 100644 --- a/web/src/components/pipeline/NodeForm/ScheduledPipeline.vue +++ b/web/src/components/pipeline/NodeForm/ScheduledPipeline.vue @@ -1498,9 +1498,39 @@ watch( watch( () => selectedStreamName.value, (val) => { - searchObj.data.stream.pipelineQueryStream = [val]; + if (searchObj?.data?.stream) { + searchObj.data.stream.pipelineQueryStream = [val]; + } }, ); + +// Watch for stream name changes and auto-generate query +// Fix for issue #9658: Auto-generate SELECT * query when stream changes +watch( + () => selectedStreamName.value, + (newStreamName, oldStreamName) => { + if (newStreamName && oldStreamName && oldStreamName !== newStreamName) { + // Stream changed: Generate new SELECT * query for the new stream + if (tab.value === "sql") { + query.value = `SELECT * FROM "${newStreamName}"`; + updateQueryValue(query.value); + } else if (tab.value === "promql") { + query.value = `${newStreamName}{}`; + updateQueryValue(query.value); + } + } else if (!oldStreamName && newStreamName) { + // Initial stream selection: Generate default query + if (tab.value === "sql" && !query.value.trim()) { + query.value = `SELECT * FROM "${newStreamName}"`; + updateQueryValue(query.value); + } else if (tab.value === "promql" && !query.value.trim()) { + query.value = `${newStreamName}{}`; + updateQueryValue(query.value); + } + } + } +); + watch( () => triggerData.value.frequency_type, (val) => { @@ -2053,13 +2083,9 @@ const getStreamFields = () => { }); }) .finally(() => { - // Only set default query if query is empty - // Don't overwrite user's custom query when they change streams - if (tab.value === "sql" && !query.value.trim()) { - query.value = `SELECT * FROM "${selectedStreamName.value}"`; - } else if (tab.value === "promql" && !query.value.trim()) { - query.value = `${selectedStreamName.value}{}`; - } + // Note: Default query generation removed + // Query is now cleared when stream changes (see watch on selectedStreamName) + // Initial query generation happens in onMounted expandState.value.query = true; expandState.value.output = false; resolve(true);