Cache, cancel and simplify the Grafana table-model metadata lookups

connectors/grafana-plugin/README.md

@@ -149,7 +149,7 @@ Unlike the tree-model modes, which go through the REST service, this mode connec
 
 DATABASE: the database to run the statement against, picked from the databases the server lists (a name can also be typed, for example a `$database` template variable). Required — every query first runs `USE <database>` on its pooled session, so results stay deterministic regardless of which session the pool hands out. Fully-qualified `database.table` references are still allowed and take precedence over the session database.
 
-TABLE (optional): picking a table lists its columns grouped by category (TAG, FIELD, ATTRIBUTE). **Use starter query** then replaces the SQL with a query that selects the table's time, TAG and numeric FIELD columns (every FIELD when none is numeric) within the dashboard time range, for example `SELECT "time", "region", "temperature" FROM "weather" WHERE $__timeFilter("time")`. Names are always double-quoted, so reserved words such as `order` and names with spaces stay valid. The table only drives this helper; the SQL is what runs.
+TABLE (optional): picking a table lists its columns grouped by category (TAG, FIELD, ATTRIBUTE). **Use starter query** then replaces the SQL with a query that selects the table's time, TAG and numeric FIELD columns (every FIELD when none is numeric) within the dashboard time range, for example `SELECT "time", "region", "temperature" FROM "weather" WHERE $__timeFilter("time")`. Names are always double-quoted, so reserved words such as `order` and names with spaces stay valid. The table only drives this helper; the SQL is what runs. The editor reuses the database, table and column lists it has read for 30 seconds, so a table created in the meantime can take that long to appear (its name can still be typed); opening the DATABASE list always reads the databases again.
 
 SQL input box: a single table-model SELECT statement. The following macros are expanded by the plugin before the statement is sent:
 

connectors/grafana-plugin/pkg/plugin/plugin.go

@@ -110,6 +110,11 @@ type IoTDBDataSource struct {
 	// fake to exercise the variable-query path without a live server.
 	tableExecutor func(ctx context.Context, database, sql string) (*tableQueryDataSet, error)
 
+	// tableSessionSource hands out a table-model session for the real RPC
+	// executor. A nil value selects the session pool; tests substitute a fake to
+	// observe how a statement uses its session without a live server.
+	tableSessionSource func() (client.ITableSession, error)
+
 	// now returns the current instant used to compute the node-liveness window
 	// for table-model template-variable queries. A nil value selects time.Now;
 	// tests substitute a fixed clock so the window is deterministic.

connectors/grafana-plugin/pkg/plugin/table_metadata.go

@@ -20,6 +20,8 @@
 package plugin
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"net/http"
 	"strings"
@@ -30,6 +32,10 @@ import (
 // Resource handlers behind the table-model query editor's database, table and
 // column pickers. Each one runs a fixed metadata statement on a pooled native
 // session; request parameters only reach the server as quoted identifiers.
+// The statements name their database themselves (SHOW TABLES FROM <db>,
+// DESC <db>.<table>), so a lookup issues no USE and changes no session state.
+// The request context is honoured as far as the native client allows; see
+// executeTableStatement.
 
 // tableColumn describes one column of a table-model table as DESC reports it.
 type tableColumn struct {
@@ -60,13 +66,13 @@ func (d *IoTDBDataSource) getTableDatabases() http.Handler {
 func (d *IoTDBDataSource) getTableTables() http.Handler {
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		// The editor trims names when they are entered; here a blank name is only
-		// rejected, and the name is used as given.
+		// rejected. The database is trimmed as a panel query's USE trims it.
 		database := r.URL.Query().Get("database")
 		if strings.TrimSpace(database) == "" {
 			writeJSONError(w, http.StatusBadRequest, "the database parameter is required")
 			return
 		}
-		dataSet, err := d.executeTableStatement(r.Context(), database, "SHOW TABLES")
+		dataSet, err := d.executeTableStatement(r.Context(), "", showTablesStatement(database))
 		if err != nil {
 			writeMetadataError(w, "SHOW TABLES", err)
 			return
@@ -90,7 +96,7 @@ func (d *IoTDBDataSource) getTableColumns() http.Handler {
 			writeJSONError(w, http.StatusBadRequest, "the database and table parameters are required")
 			return
 		}
-		dataSet, err := d.executeTableStatement(r.Context(), database, "DESC "+quoteTableIdentifier(table))
+		dataSet, err := d.executeTableStatement(r.Context(), "", describeTableStatement(database, table))
 		if err != nil {
 			writeMetadataError(w, "DESC", err)
 			return
@@ -104,8 +110,25 @@ func (d *IoTDBDataSource) getTableColumns() http.Handler {
 	})
 }
 
+// showTablesStatement lists the tables of a database without a USE.
+func showTablesStatement(database string) string {
+	return "SHOW TABLES FROM " + quoteTableIdentifier(strings.TrimSpace(database))
+}
+
+// describeTableStatement describes a table by its qualified name, without a
+// USE. The table name is used as given: the editor trims it when it is
+// entered, and the starter query names the same table.
+func describeTableStatement(database, table string) string {
+	return "DESC " + quoteTableIdentifier(strings.TrimSpace(database)) + "." + quoteTableIdentifier(table)
+}
+
 func writeMetadataError(w http.ResponseWriter, statement string, err error) {
-	log.DefaultLogger.Error("table-model metadata query failed", "statement", statement, "err", err)
+	if errors.Is(err, context.Canceled) {
+		// The editor cancels a lookup it no longer needs; that is not a failure.
+		log.DefaultLogger.Debug("table-model metadata query cancelled", "statement", statement)
+	} else {
+		log.DefaultLogger.Error("table-model metadata query failed", "statement", statement, "err", err)
+	}
 	writeJSONError(w, http.StatusInternalServerError, err.Error())
 }
 

connectors/grafana-plugin/pkg/plugin/table_metadata_test.go

@@ -28,6 +28,8 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+
+	"github.com/apache/iotdb-client-go/v2/client"
 )
 
 type recordedStatement struct {
@@ -80,7 +82,7 @@ func TestTableDatabasesReturnsTheDatabaseColumn(t *testing.T) {
 	}
 }
 
-func TestTableTablesRunsShowTablesInTheRequestedDatabase(t *testing.T) {
+func TestTableTablesNamesTheRequestedDatabaseInTheStatement(t *testing.T) {
 	var calls []recordedStatement
 	d := metadataDataSource(&tableQueryDataSet{
 		ColumnNames: []string{"TableName", "TTL(ms)"},
@@ -99,7 +101,8 @@ func TestTableTablesRunsShowTablesInTheRequestedDatabase(t *testing.T) {
 	if want := []string{"weather station", "cpu"}; !reflect.DeepEqual(names, want) {
 		t.Fatalf("names = %v, want %v", names, want)
 	}
-	if want := []recordedStatement{{database: "grafana_demo", sql: "SHOW TABLES"}}; !reflect.DeepEqual(calls, want) {
+	// SHOW TABLES FROM names the database, so no USE is needed on the session.
+	if want := []recordedStatement{{database: "", sql: `SHOW TABLES FROM "grafana_demo"`}}; !reflect.DeepEqual(calls, want) {
 		t.Fatalf("statements = %v, want %v", calls, want)
 	}
 }
@@ -134,27 +137,77 @@ func TestTableColumnsQuotesTheTableName(t *testing.T) {
 	if !reflect.DeepEqual(columns, want) {
 		t.Fatalf("columns = %v, want %v", columns, want)
 	}
-	// An embedded double quote is doubled, so the name stays one identifier.
-	if want := []recordedStatement{{database: "grafana_demo", sql: `DESC "weather""station"`}}; !reflect.DeepEqual(calls, want) {
+	// An embedded double quote is doubled, so the name stays one identifier, and
+	// the table is qualified with its database instead of a USE.
+	if want := []recordedStatement{{database: "", sql: `DESC "grafana_demo"."weather""station"`}}; !reflect.DeepEqual(calls, want) {
 		t.Fatalf("statements = %v, want %v", calls, want)
 	}
 }
 
+func TestTableMetadataQuotesEachPartOfTheName(t *testing.T) {
+	for _, tc := range []struct {
+		name    string
+		handler func(d *IoTDBDataSource) http.Handler
+		target  string
+		want    string
+	}{
+		{
+			name:    "hyphen and upper case in the database",
+			handler: (*IoTDBDataSource).getTableTables,
+			target:  "/tableTables?database=grafana-Demo",
+			want:    `SHOW TABLES FROM "grafana-Demo"`,
+		},
+		{
+			name:    "double quote in the database",
+			handler: (*IoTDBDataSource).getTableTables,
+			target:  `/tableTables?database=we%22ird`,
+			want:    `SHOW TABLES FROM "we""ird"`,
+		},
+		{
+			name:    "hyphen and upper case in both parts",
+			handler: (*IoTDBDataSource).getTableColumns,
+			target:  "/tableColumns?database=grafana-Demo&table=Weather-Station",
+			want:    `DESC "grafana-Demo"."Weather-Station"`,
+		},
+		{
+			// A dot inside a name stays inside its quoted part, so it cannot
+			// redirect the lookup to another database.
+			name:    "dot and double quote inside the names",
+			handler: (*IoTDBDataSource).getTableColumns,
+			target:  `/tableColumns?database=a.b&table=c%22.%22d`,
+			want:    `DESC "a.b"."c"".""d"`,
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var calls []recordedStatement
+			d := metadataDataSource(&tableQueryDataSet{ColumnNames: []string{"TableName", "ColumnName", "DataType", "Category"}}, nil, &calls)
+
+			if recorder := serveMetadata(t, tc.handler(d), tc.target); recorder.Code != http.StatusOK {
+				t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String())
+			}
+			if want := []recordedStatement{{database: "", sql: tc.want}}; !reflect.DeepEqual(calls, want) {
+				t.Fatalf("statements = %v, want %v", calls, want)
+			}
+		})
+	}
+}
+
 func TestTableColumnsUsesTheNamesAsGiven(t *testing.T) {
 	var calls []recordedStatement
 	d := metadataDataSource(&tableQueryDataSet{
 		ColumnNames: []string{"ColumnName", "DataType", "Category"},
 		Values:      [][]interface{}{{"time", "TIMESTAMP", "TIME"}},
 	}, nil, &calls)
 
-	recorder := serveMetadata(t, d.getTableColumns(), `/tableColumns?database=grafana_demo&table=weather%20`)
+	recorder := serveMetadata(t, d.getTableColumns(), `/tableColumns?database=%20grafana_demo%20&table=weather%20`)
 
 	if recorder.Code != http.StatusOK {
 		t.Fatalf("status = %d, body = %s", recorder.Code, recorder.Body.String())
 	}
 	// The editor trims names when they are entered, so the handler does not trim
-	// again: the columns it lists are those of the table the starter query names.
-	if want := []recordedStatement{{database: "grafana_demo", sql: `DESC "weather "`}}; !reflect.DeepEqual(calls, want) {
+	// the table again: the columns it lists are those of the table the starter
+	// query names. The database is trimmed, as a panel query's USE trims it.
+	if want := []recordedStatement{{database: "", sql: `DESC "grafana_demo"."weather "`}}; !reflect.DeepEqual(calls, want) {
 		t.Fatalf("statements = %v, want %v", calls, want)
 	}
 }
@@ -181,6 +234,33 @@ func TestTableMetadataRequiresItsParameters(t *testing.T) {
 	}
 }
 
+func TestTableMetadataStopsWhenTheEditorCancelsTheRequest(t *testing.T) {
+	acquired := 0
+	d := sessionDataSource(&acquired, func() client.ITableSession { return &fakeTableSession{} })
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	for _, tc := range []struct {
+		handler http.Handler
+		target  string
+	}{
+		{d.getTableDatabases(), "/tableDatabases"},
+		{d.getTableTables(), "/tableTables?database=grafana_demo"},
+		{d.getTableColumns(), "/tableColumns?database=grafana_demo&table=cpu"},
+	} {
+		recorder := httptest.NewRecorder()
+		tc.handler.ServeHTTP(recorder, httptest.NewRequest(http.MethodGet, tc.target, nil).WithContext(ctx))
+		if recorder.Code != http.StatusInternalServerError {
+			t.Errorf("%s: status = %d, want %d", tc.target, recorder.Code, http.StatusInternalServerError)
+		}
+	}
+	// The handlers pass the request's context on, so an aborted lookup never
+	// takes one of the pool's sessions from the panel queries.
+	if acquired != 0 {
+		t.Fatalf("cancelled lookups took %d session(s)", acquired)
+	}
+}
+
 func TestTableMetadataReportsServerErrors(t *testing.T) {
 	var calls []recordedStatement
 	d := metadataDataSource(nil, errors.New("550: Table 'grafana_demo.nope' does not exist"), &calls)

connectors/grafana-plugin/pkg/plugin/table_query.go

@@ -245,7 +245,8 @@ func formatIoTDBDuration(intervalMS int64) (string, error) {
 
 // quoteTableIdentifier wraps a table-model identifier in double quotes
 // (doubling any embedded quote), the relational grammar's quoted-identifier
-// form, so a database name survives the USE statement verbatim.
+// form, so a database or table name reaches the server verbatim. Each part of
+// a qualified name is quoted on its own ("db"."table").
 func quoteTableIdentifier(name string) string {
 	return `"` + strings.ReplaceAll(name, `"`, `""`) + `"`
 }
@@ -521,19 +522,34 @@ func instantSeriesKey(dataSet *tableQueryDataSet, row []interface{}, timeCol int
 
 // executeTableStatement runs a table-model SQL statement against the given
 // database on a pooled native-client session and returns the fetched dataset.
-// The database is USEd on the session so a statement's table references resolve
-// against it regardless of any session state left by earlier queries. When
-// tableExecutor is non-nil it is used instead of the real RPC path.
+// A non-empty database is USEd on the session so a statement's table
+// references resolve against it regardless of any session state left by
+// earlier queries; a statement that names its database itself passes an empty
+// one and issues no USE. When tableExecutor is non-nil it is used instead of
+// the real RPC path.
+//
+// The native client's pool takes no context: GetSession can wait up to
+// tableWaitSessionTimeoutMs for a free session and cannot be interrupted. The
+// context is therefore checked before that wait and again after it, so a
+// request that has been cancelled, or whose deadline has passed, never runs a
+// statement whose result nobody will read.
 func (d *IoTDBDataSource) executeTableStatement(ctx context.Context, database, sql string) (*tableQueryDataSet, error) {
 	if d.tableExecutor != nil {
 		return d.tableExecutor(ctx, database, sql)
 	}
 
-	pool, err := d.getTablePool()
-	if err != nil {
+	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
-	session, err := pool.GetSession()
+	getSession := d.tableSessionSource
+	if getSession == nil {
+		pool, err := d.getTablePool()
+		if err != nil {
+			return nil, err
+		}
+		getSession = pool.GetSession
+	}
+	session, err := getSession()
 	if err != nil {
 		log.DefaultLogger.Error("Cannot connect to the IoTDB RPC service", "err", err)
 		return nil, fmt.Errorf("cannot connect to the IoTDB RPC service: %w", err)
@@ -543,6 +559,11 @@ func (d *IoTDBDataSource) executeTableStatement(ctx context.Context, database, s
 			log.DefaultLogger.Error("Failed to return session to the pool", "err", closeErr)
 		}
 	}()
+	if err := ctx.Err(); err != nil {
+		// The caller gave up while this request waited for a session; hand the
+		// session straight back to the pool for the requests still waiting.
+		return nil, err
+	}
 
 	if db := strings.TrimSpace(database); db != "" {
 		if err := session.ExecuteNonQueryStatement("USE " + quoteTableIdentifier(db)); err != nil {