Skip to content

Remove cluster check for entity type #9713

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
k-anshul merged 5 commits into from
Jul 21, 2026
Merged

Remove cluster check for entity type #9713

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
6e9105a
use view instead of directly querying system.tables in clusterAllRepl…
k-anshul Jul 20, 2026
24845d8
tests fixes
k-anshul Jul 20, 2026
d093f44
review
k-anshul Jul 20, 2026
19bc2c6
nit
k-anshul Jul 20, 2026
7eccde1
build fix
k-anshul Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 6 additions & 6 deletions runtime/drivers/clickhouse/clickhouse.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -860,14 +860,14 @@ func (c *Connection) checkBillingTableExists(ctx context.Context, cluster string
return false, fmt.Errorf("failed to check if billing table exists: %w", err)
}
} else {
// Query through view so ClickHouse applies normal SELECT access checks on system.tables.
// Targeting system.tables directly requires SHOW COLUMNS on system.tables starting in ClickHouse 26.3.
err := c.readDB.QueryRowxContext(ctx, fmt.Sprintf(`
SELECT countIf(found) = count() AS exists_everywhere, countIf(found) > 0 AS exists_somewhere
FROM
(
SELECT hostName() AS host, max((database = 'billing') AND (name = 'events')) AS found
FROM clusterAllReplicas('%s', system.tables)
GROUP BY host
)`, cluster)).Scan(&existsEverywhere, &existsSomewhere)
FROM clusterAllReplicas(%s, view(
SELECT max((database = 'billing') AND (name = 'events')) AS found
FROM system.tables
))`, safeSQLString(cluster))).Scan(&existsEverywhere, &existsSomewhere)
if err != nil {
return false, fmt.Errorf("failed to check if billing table exists in cluster %q: %w", cluster, err)
}
Expand Down
80 changes: 32 additions & 48 deletions runtime/drivers/clickhouse/crud.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,10 +21,7 @@ type tableWriteMetrics struct {
}

func (c *Connection) createTableAsSelect(ctx context.Context, name, sql string, outputProps *ModelOutputProperties, beforeCreate, afterCreate string) (*tableWriteMetrics, error) {
var onClusterClause string
if c.config.Cluster != "" {
onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster)
}
onClusterClause := c.onClusterClause()

// Execute beforeCreate query
if beforeCreate != "" {
Expand Down Expand Up @@ -116,14 +113,7 @@ func (c *Connection) insertTableAsSelect(ctx context.Context, name, sql string,
}

if opts.Strategy == drivers.IncrementalStrategyPartitionOverwrite {
_, onCluster, err := c.entityType(ctx, c.config.Database, name)
if err != nil {
return nil, err
}
onClusterClause := ""
if onCluster {
onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster)
}
onClusterClause := c.onClusterClause()
// Get the engine info of the given table
engine, err := c.getTableEngine(ctx, name)
if err != nil {
Expand Down Expand Up @@ -210,13 +200,9 @@ func (c *Connection) insertTableAsSelect(ctx context.Context, name, sql string,
}

if opts.Strategy == drivers.IncrementalStrategyMerge {
_, onCluster, err := c.entityType(ctx, c.config.Database, name)
if err != nil {
return nil, err
}
// get the engine info of the given table - local table for distributed tables
var n string
if onCluster {
if c.config.Cluster != "" {
n = localTableName(name)
} else {
n = name
Expand Down Expand Up @@ -249,45 +235,46 @@ func (c *Connection) insertTableAsSelect(ctx context.Context, name, sql string,
}

func (c *Connection) dropTable(ctx context.Context, name string) error {
typ, onCluster, err := c.entityType(ctx, c.config.Database, name)
typ, err := c.entityType(ctx, c.config.Database, name)
if err != nil {
return err
}
var onClusterClause string
if onCluster {
onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster)
}

onClusterClause := c.onClusterClause()

switch typ {
case "VIEW":
return c.Exec(ctx, &drivers.Statement{
Comment thread
k-anshul marked this conversation as resolved.
Query: fmt.Sprintf("DROP VIEW %s %s", safeSQLName(name), onClusterClause),
Query: fmt.Sprintf("DROP VIEW IF EXISTS %s %s", safeSQLName(name), onClusterClause),
Priority: 100,
})
case "DICTIONARY":
// first drop the dictionary
err := c.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("DROP DICTIONARY %s %s", safeSQLName(name), onClusterClause),
Query: fmt.Sprintf("DROP DICTIONARY IF EXISTS %s %s", safeSQLName(name), onClusterClause),
Priority: 100,
})
// then drop the temp table
_ = c.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("DROP TABLE %s %s", safeSQLName(tempTableForDictionary(name)), onClusterClause),
Query: fmt.Sprintf("DROP TABLE IF EXISTS %s %s", safeSQLName(tempTableForDictionary(name)), onClusterClause),
Priority: 100,
})
return err
case "TABLE":
// drop the main table
// use IF EXISTS so drops succeed in cluster mode even for tables that don't exist on every node,
// e.g. tables created before the cluster was configured or without a `_local` counterpart
err := c.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("DROP TABLE %s %s", safeSQLName(name), onClusterClause),
Query: fmt.Sprintf("DROP TABLE IF EXISTS %s %s", safeSQLName(name), onClusterClause),
Priority: 100,
})
if err != nil {
return err
}
// then drop the local table in case of cluster
if onCluster && !strings.HasSuffix(name, "_local") {
if c.config.Cluster != "" && !strings.HasSuffix(name, "_local") {
return c.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("DROP TABLE %s %s", safeSQLName(localTableName(name)), onClusterClause),
Query: fmt.Sprintf("DROP TABLE IF EXISTS %s %s", safeSQLName(localTableName(name)), onClusterClause),
Priority: 100,
})
}
Expand All @@ -298,22 +285,19 @@ func (c *Connection) dropTable(ctx context.Context, name string) error {
}

func (c *Connection) renameEntity(ctx context.Context, oldName, newName string) error {
typ, onCluster, err := c.entityType(ctx, c.config.Database, oldName)
typ, err := c.entityType(ctx, c.config.Database, oldName)
if err != nil {
return err
}
var onClusterClause string
if onCluster {
onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster)
}
onClusterClause := c.onClusterClause()

switch typ {
case "VIEW":
return c.renameView(ctx, oldName, newName, onClusterClause)
case "DICTIONARY":
return c.renameTable(ctx, oldName, newName, onClusterClause)
case "TABLE":
if !onCluster {
if c.config.Cluster == "" {
return c.renameTable(ctx, oldName, newName, onClusterClause)
}
// capture the full engine of the old distributed table
Expand Down Expand Up @@ -447,10 +431,7 @@ func (c *Connection) renameTable(ctx context.Context, oldName, newName, onCluste
}

func (c *Connection) createTable(ctx context.Context, name, sql string, outputProps *ModelOutputProperties) error {
var onClusterClause string
if c.config.Cluster != "" {
onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster)
}
onClusterClause := c.onClusterClause()
var create strings.Builder
create.WriteString("CREATE OR REPLACE TABLE ")
if c.config.Cluster != "" {
Expand Down Expand Up @@ -512,7 +493,7 @@ func (c *Connection) createDistributedTable(ctx context.Context, name string, ou
if c.config.Cluster == "" {
return fmt.Errorf("clickhouse: cannot create distributed table without a cluster")
}
onClusterClause := "ON CLUSTER " + safeSQLName(c.config.Cluster)
onClusterClause := c.onClusterClause()

var distributed strings.Builder
database := "currentDatabase()"
Expand All @@ -534,10 +515,7 @@ func (c *Connection) createDistributedTable(ctx context.Context, name string, ou
}

func (c *Connection) createDictionary(ctx context.Context, name, sql string, outputProps *ModelOutputProperties) error {
var onClusterClause string
if c.config.Cluster != "" {
onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster)
}
onClusterClause := c.onClusterClause()
if sql == "" {
if outputProps.Columns == "" {
return fmt.Errorf("clickhouse: no columns specified for dictionary %q", name)
Expand Down Expand Up @@ -642,7 +620,7 @@ func (c *Connection) getTableEngine(ctx context.Context, name string) (string, e
return "", err
}
defer res.Close()
if res.Next() {
for res.Next() {
if err := res.Scan(&engine); err != nil {
return "", err
}
Expand Down Expand Up @@ -695,9 +673,8 @@ func (c *Connection) getTablePartitions(ctx context.Context, name string) ([]str
}

func (c *Connection) replacePartition(ctx context.Context, src, dest, part string) error {
var onClusterClause string
onClusterClause := c.onClusterClause()
if c.config.Cluster != "" {
onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster)
dest = localTableName(dest)
src = localTableName(src)
}
Expand All @@ -718,9 +695,8 @@ func (c *Connection) syncReplica(ctx context.Context, tableName string) error {
if err != nil {
return err
}
onClusterClause := "ON CLUSTER " + safeSQLName(c.config.Cluster)
return c.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("SYSTEM SYNC REPLICA %s %s.%s", onClusterClause, safeSQLName(database), safeSQLName(localTableName(tableName))),
Query: fmt.Sprintf("SYSTEM SYNC REPLICA %s %s.%s", c.onClusterClause(), safeSQLName(database), safeSQLName(localTableName(tableName))),
Priority: 1,
})
}
Expand Down Expand Up @@ -750,6 +726,14 @@ func (c *Connection) currentDatabase(ctx context.Context) (string, error) {
return database, nil
}

// onClusterClause returns the ON CLUSTER clause for DDL statements, or an empty string if no cluster is configured.
func (c *Connection) onClusterClause() string {
if c.config.Cluster == "" {
return ""
}
return "ON CLUSTER " + safeSQLName(c.config.Cluster)
}

func isReplicatedEngine(engine string) bool {
return strings.Contains(strings.ToLower(engine), "replicated")
}
Expand Down
35 changes: 11 additions & 24 deletions runtime/drivers/clickhouse/information_schema.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -474,43 +474,30 @@ func scanTables(rows *sqlx.Rows) ([]*drivers.OlapTable, error) {
return res, nil
}

func (c *Connection) entityType(ctx context.Context, db, name string) (typ string, onCluster bool, err error) {
func (c *Connection) entityType(ctx context.Context, db, name string) (typ string, err error) {
conn, release, err := c.acquireMetaConn(ctx)
if err != nil {
return "", false, err
return "", err
}
defer func() { _ = release() }()

var q string
if c.config.Cluster == "" {
q = `SELECT
multiIf(engine IN ('MaterializedView', 'View'), 'VIEW', engine = 'Dictionary', 'DICTIONARY', 'TABLE') AS type,
0 AS is_on_cluster
FROM system.tables AS t
JOIN system.databases AS db ON t.database = db.name
WHERE t.database = coalesce(?, currentDatabase()) AND t.name = ?`
} else {
q = `SELECT
multiIf(engine IN ('MaterializedView', 'View'), 'VIEW', engine = 'Dictionary', 'DICTIONARY', 'TABLE') AS type,
countDistinct(_shard_num) > 1 AS is_on_cluster
FROM clusterAllReplicas(` + safeSQLName(c.config.Cluster) + `, system.tables) AS t
JOIN system.databases AS db ON t.database = db.name
WHERE t.database = coalesce(?, currentDatabase()) AND t.name = ?
GROUP BY engine, t.name`
}
// A local system.tables lookup suffices even when a cluster is configured:
// all DDL in cluster mode runs ON CLUSTER, so the entity always exists on the connected node.
q := `SELECT multiIf(engine IN ('MaterializedView', 'View'), 'VIEW', engine = 'Dictionary', 'DICTIONARY', 'TABLE') AS type
FROM system.tables
WHERE database = coalesce(?, currentDatabase()) AND name = ?`
Comment on lines +484 to +488

@chatgpt-codex-connector chatgpt-codex-connector Bot Jul 20, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Preserve local handling for pre-cluster tables

When cluster mode is enabled for an existing deployment, tables created before that configuration can exist only under their original name and have no <name>_local table. This lookup now discards the previous per-entity cluster determination, so renameEntity always follows the distributed-table path and attempts to rename <old>_local on every cluster node (see crud.go:300-338), causing normal model rebuild/rename operations to fail for those existing tables. The table-drop comment already identifies this supported migration scenario; retain per-entity detection or handle legacy local tables separately.

Useful? React with 👍 / 👎.

@k-anshul k-anshul Jul 20, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IMO it is fine to exclude this scenario. The callers of renameEntity are :

  1. When staging tables are renamed. The migration from single node to cluster can't happen in that case.
  2. When models are renamed. Again a very edge case where model is renamed and migration from single node to cluster is also done.

For migration it is recommended to rebuild full models.

var args []any
if db == "" {
args = []any{nil, name}
} else {
args = []any{db, name}
}
row := conn.QueryRowxContext(ctx, q, args...)
err = row.Scan(&typ, &onCluster)
err = conn.QueryRowxContext(ctx, q, args...).Scan(&typ)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return "", false, drivers.ErrNotFound
return "", drivers.ErrNotFound
}
return "", false, err
return "", err
}
return typ, onCluster, nil
return typ, nil
}
65 changes: 65 additions & 0 deletions runtime/drivers/clickhouse/olap_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@ package clickhouse
import (
"context"
"fmt"
"strings"
"testing"

runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1"
Expand All @@ -29,6 +30,7 @@ func TestClickhouseSingle(t *testing.T) {
require.True(t, ok)

t.Run("WithConnection", func(t *testing.T) { testWithConnection(t, olap) })
t.Run("DropTable", func(t *testing.T) { testDropTable(t, c, olap) })
t.Run("RenameView", func(t *testing.T) { testRenameView(t, c, olap) })
t.Run("RenameTable", func(t *testing.T) { testRenameTable(t, c, olap) })
t.Run("CreateTableAsSelect", func(t *testing.T) { testCreateTableAsSelect(t, c) })
Expand Down Expand Up @@ -59,6 +61,7 @@ func TestClickhouseCluster(t *testing.T) {
prepareClusterConn(t, olap, cluster)

t.Run("WithConnection", func(t *testing.T) { testWithConnection(t, olap) })
t.Run("DropTable", func(t *testing.T) { testDropTable(t, c, olap) })
t.Run("RenameView", func(t *testing.T) { testRenameView(t, c, olap) })
t.Run("RenameTable", func(t *testing.T) { testRenameTable(t, c, olap) })
t.Run("CreateTableAsSelect", func(t *testing.T) { testCreateTableAsSelect(t, c) })
Expand All @@ -69,6 +72,8 @@ func TestClickhouseCluster(t *testing.T) {
t.Run("SyncReplica_NonDefaultDatabase", func(t *testing.T) { testSyncReplicaNonDefaultDatabase(t, olap, dsn, cluster) })
t.Run("TestDictionary", func(t *testing.T) { testDictionary(t, c, olap) })
t.Run("QueryAttributesAsSettings", func(t *testing.T) { testQueryAttributesAsSettings(t, olap) })
t.Run("EntityTypeRestrictedUser", func(t *testing.T) { testEntityTypeRestrictedUser(t, olap, dsn, cluster) })

}

func testWithConnection(t *testing.T, olap drivers.OLAPStore) {
Expand Down Expand Up @@ -98,6 +103,23 @@ func testWithConnection(t *testing.T, olap drivers.OLAPStore) {
require.NoError(t, err)
}

func testDropTable(t *testing.T, c *Connection, olap drivers.OLAPStore) {
ctx := context.Background()

_, err := c.createTableAsSelect(ctx, "drop_table", "SELECT 1 AS id", &ModelOutputProperties{Engine: "MergeTree"}, "", "")
require.NoError(t, err)
require.NoError(t, c.dropTable(ctx, "drop_table"))
notExists(t, olap, "drop_table")
if c.config.Cluster != "" {
notExists(t, olap, localTableName("drop_table"))
}

_, err = c.createTableAsSelect(ctx, "drop_view", "SELECT 1 AS id", &ModelOutputProperties{Typ: "VIEW"}, "", "")
require.NoError(t, err)
require.NoError(t, c.dropTable(ctx, "drop_view"))
notExists(t, olap, "drop_view")
}

func testRenameView(t *testing.T, c *Connection, olap drivers.OLAPStore) {
ctx := context.Background()
opts := &ModelOutputProperties{Typ: "VIEW"}
Expand Down Expand Up @@ -1027,3 +1049,46 @@ func testQueryAttributesAsSettings(t *testing.T, olap drivers.OLAPStore) {
require.Contains(t, err.Error(), "neither a builtin setting nor")
})
}

func testEntityTypeRestrictedUser(t *testing.T, olap drivers.OLAPStore, dsn, cluster string) {
ctx := context.Background()
const (
username = "rill_restricted"
password = "test_password"
)

require.NoError(t, olap.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("CREATE USER %s ON CLUSTER %s IDENTIFIED WITH sha256_password BY %s", safeSQLName(username), safeSQLName(cluster), safeSQLString(password)),
}))
t.Cleanup(func() {
_ = olap.Exec(context.Background(), &drivers.Statement{Query: fmt.Sprintf("DROP USER IF EXISTS %s ON CLUSTER %s", safeSQLName(username), safeSQLName(cluster))})
})
require.NoError(t, olap.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("GRANT ON CLUSTER %s SELECT ON default.* TO %s", safeSQLName(cluster), safeSQLName(username)),
}))
require.NoError(t, olap.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("GRANT ON CLUSTER %s CLUSTER, REMOTE ON *.* TO %s", safeSQLName(cluster), safeSQLName(username)),
}))

// create a dedicated table: other subtests rename or drop the shared tables created by prepareClusterConn
require.NoError(t, olap.Exec(ctx, &drivers.Statement{
Query: fmt.Sprintf("CREATE OR REPLACE TABLE entity_type_restricted ON CLUSTER %s (x Int32) engine=MergeTree ORDER BY x", safeSQLName(cluster)),
}))
t.Cleanup(func() {
_ = olap.Exec(context.Background(), &drivers.Statement{Query: fmt.Sprintf("DROP TABLE IF EXISTS entity_type_restricted ON CLUSTER %s", safeSQLName(cluster))})
})

restrictedDSN := strings.Replace(dsn, "clickhouse://default@", fmt.Sprintf("clickhouse://%s:%s@", username, password), 1)
handle, err := drivers.Open("clickhouse", "", "restricted", map[string]any{"dsn": restrictedDSN, "cluster": cluster}, storage.MustNew(t.TempDir(), nil), activity.NewNoopClient(), zap.NewNop())
require.NoError(t, err)
defer handle.Close()

restrictedConnection := handle.(*Connection)
typ, err := restrictedConnection.entityType(ctx, "default", "entity_type_restricted")
require.NoError(t, err)
require.Equal(t, "TABLE", typ)

exists, err := restrictedConnection.checkBillingTableExists(ctx, cluster)
require.NoError(t, err)
require.False(t, exists)
}
2 changes: 1 addition & 1 deletion runtime/drivers/clickhouse/testclickhouse/testclickhouse.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ func Start(t TestingT) string {
ctx := context.Background()
clickHouseContainer, err := clickhouse.Run(
ctx,
"clickhouse/clickhouse-server:25.6.12.10",
"clickhouse/clickhouse-server:26.3.1.896",
clickhouse.WithConfigFile(filepath.Join(testdataPath, "clickhouse-config.xml")),
testcontainers.CustomizeRequestOption(func(req *testcontainers.GenericContainerRequest) error {
cf := testcontainers.ContainerFile{
Expand Down
Loading
Loading