From 6e9105a0ea1c815c8de70ca8e6c86b78027b3503 Mon Sep 17 00:00:00 2001 From: Anshul Khandelwal <12948312+k-anshul@users.noreply.github.com> Date: Mon, 20 Jul 2026 14:48:28 +0530 Subject: [PATCH 1/5] use view instead of directly querying system.tables in clusterAllReplicas --- runtime/drivers/clickhouse/clickhouse.go | 12 ++-- runtime/drivers/clickhouse/crud.go | 26 ++++----- .../drivers/clickhouse/information_schema.go | 38 ++++++------- runtime/drivers/clickhouse/olap_test.go | 57 +++++++++++++++++++ .../testclickhouse/testclickhouse.go | 2 +- .../ch_cluster_2S_2R/docker-compose.yaml | 10 ++-- 6 files changed, 98 insertions(+), 47 deletions(-) diff --git a/runtime/drivers/clickhouse/clickhouse.go b/runtime/drivers/clickhouse/clickhouse.go index 50da5eb6ff87..ad76749780cf 100644 --- a/runtime/drivers/clickhouse/clickhouse.go +++ b/runtime/drivers/clickhouse/clickhouse.go @@ -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) } diff --git a/runtime/drivers/clickhouse/crud.go b/runtime/drivers/clickhouse/crud.go index 988412321117..f740ac13f53e 100644 --- a/runtime/drivers/clickhouse/crud.go +++ b/runtime/drivers/clickhouse/crud.go @@ -116,12 +116,8 @@ 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 { + if c.config.Cluster != "" { onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster) } // Get the engine info of the given table @@ -210,13 +206,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 @@ -249,14 +241,16 @@ 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 { + if c.config.Cluster != "" { onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster) } + switch typ { case "VIEW": return c.Exec(ctx, &drivers.Statement{ @@ -285,7 +279,7 @@ func (c *Connection) dropTable(ctx context.Context, name string) error { 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), Priority: 100, @@ -298,12 +292,12 @@ 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 { + if c.config.Cluster != "" { onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster) } @@ -313,7 +307,7 @@ func (c *Connection) renameEntity(ctx context.Context, oldName, newName string) 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 diff --git a/runtime/drivers/clickhouse/information_schema.go b/runtime/drivers/clickhouse/information_schema.go index 5e55ce23453f..3323533b1f1b 100644 --- a/runtime/drivers/clickhouse/information_schema.go +++ b/runtime/drivers/clickhouse/information_schema.go @@ -474,29 +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 = ?` + multiIf(engine IN ('MaterializedView', 'View'), 'VIEW', engine = 'Dictionary', 'DICTIONARY', 'TABLE') AS type + 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` + // 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. + q = `SELECT type + FROM clusterAllReplicas(` + safeSQLString(c.config.Cluster) + `, view( + SELECT multiIf(engine IN ('MaterializedView', 'View'), 'VIEW', engine = 'Dictionary', 'DICTIONARY', 'TABLE') AS type + FROM system.tables + WHERE database = coalesce(?, currentDatabase()) AND name = ? + )) + GROUP BY type` } var args []any if db == "" { @@ -504,13 +505,12 @@ func (c *Connection) entityType(ctx context.Context, db, name string) (typ strin } 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 } diff --git a/runtime/drivers/clickhouse/olap_test.go b/runtime/drivers/clickhouse/olap_test.go index 7cb20510ffe9..8ea352852071 100644 --- a/runtime/drivers/clickhouse/olap_test.go +++ b/runtime/drivers/clickhouse/olap_test.go @@ -3,6 +3,7 @@ package clickhouse import ( "context" "fmt" + "strings" "testing" runtimev1 "github.com/rilldata/rill/proto/gen/rill/runtime/v1" @@ -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) }) @@ -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) }) @@ -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) { @@ -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"} @@ -1027,3 +1049,38 @@ 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)), + })) + + 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", "foo") + require.NoError(t, err) + require.Equal(t, "TABLE", typ) + + exists, err := restrictedConnection.checkBillingTableExists(ctx, cluster) + require.NoError(t, err) + require.False(t, exists) +} diff --git a/runtime/drivers/clickhouse/testclickhouse/testclickhouse.go b/runtime/drivers/clickhouse/testclickhouse/testclickhouse.go index 2e7841471993..4334edb1748a 100644 --- a/runtime/drivers/clickhouse/testclickhouse/testclickhouse.go +++ b/runtime/drivers/clickhouse/testclickhouse/testclickhouse.go @@ -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{ diff --git a/runtime/drivers/clickhouse/testclickhouse/testdata/ch_cluster_2S_2R/docker-compose.yaml b/runtime/drivers/clickhouse/testclickhouse/testdata/ch_cluster_2S_2R/docker-compose.yaml index 3a1712de04fb..2e8094467c88 100644 --- a/runtime/drivers/clickhouse/testclickhouse/testdata/ch_cluster_2S_2R/docker-compose.yaml +++ b/runtime/drivers/clickhouse/testclickhouse/testdata/ch_cluster_2S_2R/docker-compose.yaml @@ -1,7 +1,7 @@ version: '3.8' services: clickhouse-01: - image: "clickhouse/clickhouse-server:25.5.1.2782" + image: "clickhouse/clickhouse-server:26.3.1.896" user: "101:101" container_name: clickhouse-01 hostname: clickhouse-01 @@ -15,7 +15,7 @@ services: depends_on: - clickhouse-keeper-01 clickhouse-02: - image: "clickhouse/clickhouse-server:25.5.1.2782" + image: "clickhouse/clickhouse-server:26.3.1.896" user: "101:101" container_name: clickhouse-02 hostname: clickhouse-02 @@ -29,7 +29,7 @@ services: depends_on: - clickhouse-keeper-01 clickhouse-03: - image: "clickhouse/clickhouse-server:25.5.1.2782" + image: "clickhouse/clickhouse-server:26.3.1.896" user: "101:101" container_name: clickhouse-03 hostname: clickhouse-03 @@ -43,7 +43,7 @@ services: depends_on: - clickhouse-keeper-01 clickhouse-04: - image: "clickhouse/clickhouse-server:25.5.1.2782" + image: "clickhouse/clickhouse-server:26.3.1.896" user: "101:101" container_name: clickhouse-04 hostname: clickhouse-04 @@ -57,7 +57,7 @@ services: depends_on: - clickhouse-keeper-01 clickhouse-keeper-01: - image: "clickhouse/clickhouse-keeper:25.5.1.2782-alpine" + image: "clickhouse/clickhouse-keeper:26.3.1.896-alpine" user: "101:101" container_name: clickhouse-keeper-01 hostname: clickhouse-keeper-01 From 24845d81e06bfc058a61f58a8a78abb66a762ed4 Mon Sep 17 00:00:00 2001 From: Anshul Khandelwal <12948312+k-anshul@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:31:21 +0530 Subject: [PATCH 2/5] tests fixes --- runtime/drivers/clickhouse/crud.go | 68 ++++++++----------- .../drivers/clickhouse/information_schema.go | 23 ++----- runtime/drivers/clickhouse/olap_test.go | 10 ++- .../resolvers/testdata/connector_azure.yaml | 10 +-- .../testdata/connector_bigquery.yaml | 28 ++++---- runtime/resolvers/testdata/connector_gcs.yaml | 2 +- .../resolvers/testdata/connector_mysql.yaml | 8 +-- .../testdata/connector_postgres.yaml | 2 +- runtime/resolvers/testdata/connector_s3.yaml | 10 +-- 9 files changed, 73 insertions(+), 88 deletions(-) diff --git a/runtime/drivers/clickhouse/crud.go b/runtime/drivers/clickhouse/crud.go index f740ac13f53e..1d830aa28eb6 100644 --- a/runtime/drivers/clickhouse/crud.go +++ b/runtime/drivers/clickhouse/crud.go @@ -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 != "" { @@ -116,10 +113,7 @@ func (c *Connection) insertTableAsSelect(ctx context.Context, name, sql string, } if opts.Strategy == drivers.IncrementalStrategyPartitionOverwrite { - onClusterClause := "" - if c.config.Cluster != "" { - 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 { @@ -246,33 +240,32 @@ func (c *Connection) dropTable(ctx context.Context, name string) error { return err } - var onClusterClause string - if c.config.Cluster != "" { - onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster) - } + onClusterClause := c.onClusterClause() switch typ { case "VIEW": return c.Exec(ctx, &drivers.Statement{ - 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 { @@ -281,7 +274,7 @@ func (c *Connection) dropTable(ctx context.Context, name string) error { // then drop the local table in case of cluster 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, }) } @@ -296,10 +289,7 @@ func (c *Connection) renameEntity(ctx context.Context, oldName, newName string) if err != nil { return err } - var onClusterClause string - if c.config.Cluster != "" { - onClusterClause = "ON CLUSTER " + safeSQLName(c.config.Cluster) - } + onClusterClause := c.onClusterClause() switch typ { case "VIEW": @@ -441,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 != "" { @@ -506,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()" @@ -528,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) @@ -636,16 +620,16 @@ func (c *Connection) getTableEngine(ctx context.Context, name string) (string, e return "", err } defer res.Close() - if res.Next() { - if err := res.Scan(&engine); err != nil { + if !res.Next() { + if err := res.Err(); err != nil { return "", err } + return "", fmt.Errorf("clickhouse: table %q not found", name) } - err = res.Err() - if err != nil { + if err := res.Scan(&engine); err != nil { return "", err } - return engine, nil + return engine, res.Err() } func (c *Connection) getTablePartitions(ctx context.Context, name string) ([]string, error) { @@ -689,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) } @@ -712,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, }) } @@ -744,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") } diff --git a/runtime/drivers/clickhouse/information_schema.go b/runtime/drivers/clickhouse/information_schema.go index 3323533b1f1b..e47157f49254 100644 --- a/runtime/drivers/clickhouse/information_schema.go +++ b/runtime/drivers/clickhouse/information_schema.go @@ -481,24 +481,11 @@ func (c *Connection) entityType(ctx context.Context, db, name string) (typ strin } defer func() { _ = release() }() - var q string - if c.config.Cluster == "" { - q = `SELECT - multiIf(engine IN ('MaterializedView', 'View'), 'VIEW', engine = 'Dictionary', 'DICTIONARY', 'TABLE') AS type - FROM system.tables AS t - JOIN system.databases AS db ON t.database = db.name - WHERE t.database = coalesce(?, currentDatabase()) AND t.name = ?` - } 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. - q = `SELECT type - FROM clusterAllReplicas(` + safeSQLString(c.config.Cluster) + `, view( - SELECT multiIf(engine IN ('MaterializedView', 'View'), 'VIEW', engine = 'Dictionary', 'DICTIONARY', 'TABLE') AS type - FROM system.tables - WHERE database = coalesce(?, currentDatabase()) AND name = ? - )) - GROUP BY type` - } + // 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 = ?` var args []any if db == "" { args = []any{nil, name} diff --git a/runtime/drivers/clickhouse/olap_test.go b/runtime/drivers/clickhouse/olap_test.go index 8ea352852071..409dc139b9cb 100644 --- a/runtime/drivers/clickhouse/olap_test.go +++ b/runtime/drivers/clickhouse/olap_test.go @@ -1070,13 +1070,21 @@ func testEntityTypeRestrictedUser(t *testing.T, olap drivers.OLAPStore, dsn, clu 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", "foo") + typ, err := restrictedConnection.entityType(ctx, "default", "entity_type_restricted") require.NoError(t, err) require.Equal(t, "TABLE", typ) diff --git a/runtime/resolvers/testdata/connector_azure.yaml b/runtime/resolvers/testdata/connector_azure.yaml index 2104aeba6eab..185611a9de75 100644 --- a/runtime/resolvers/testdata/connector_azure.yaml +++ b/runtime/resolvers/testdata/connector_azure.yaml @@ -271,10 +271,10 @@ tests: string_col,Nullable(String) decimal_col,"Nullable(Decimal(10, 2))" date_col,Nullable(Date32) - time_millis_col,Nullable(DateTime64(3)) - time_micros_col,Nullable(DateTime64(6)) - timestamp_millis_col,Nullable(DateTime64(3)) - timestamp_micros_col,Nullable(DateTime64(6)) + time_millis_col,"Nullable(DateTime64(3, 'UTC'))" + time_micros_col,"Nullable(DateTime64(6, 'UTC'))" + timestamp_millis_col,"Nullable(DateTime64(3, 'UTC'))" + timestamp_micros_col,"Nullable(DateTime64(6, 'UTC'))" uuid_col,Nullable(FixedString(16)) list_int_col,Array(Nullable(Int32)) list_string_col,Array(Nullable(String)) @@ -308,7 +308,7 @@ tests: resolver: sql properties: connector: clickhouse - sql: "select * from all_datatypes_clickhouse_glob order by id" + sql: "select id, boolean_col, int32_col, int64_col, float_col, double_col, byte_array_col, fixed_len_byte_array_col, string_col, decimal_col, date_col, time_millis_col, time_micros_col, timestamp_millis_col, timestamp_micros_col, uuid_col, list_int_col, list_string_col, map_col, struct_col from all_datatypes_clickhouse_glob order by id" # time is converted to datetime like 2025-01-01T00:02:03.456Z but should be 1970-01-01T00:02:03.456Z, which is wrong. result_csv: | id,boolean_col,int32_col,int64_col,float_col,double_col,byte_array_col,fixed_len_byte_array_col,string_col,decimal_col,date_col,time_millis_col,time_micros_col,timestamp_millis_col,timestamp_micros_col,uuid_col,list_int_col,list_string_col,map_col,struct_col diff --git a/runtime/resolvers/testdata/connector_bigquery.yaml b/runtime/resolvers/testdata/connector_bigquery.yaml index c8da1a5565ff..30df0ca2341f 100644 --- a/runtime/resolvers/testdata/connector_bigquery.yaml +++ b/runtime/resolvers/testdata/connector_bigquery.yaml @@ -79,20 +79,20 @@ tests: string_col,Nullable(String) bytes_col,Nullable(String) date_col,Nullable(Date32) - datetime_col,Nullable(DateTime64(6)) - time_col,Nullable(DateTime64(6)) - timestamp_col,Nullable(DateTime64(6)) - array_int_col,Array(Nullable(Int64)) - array_float_col,Array(Nullable(Float64)) - array_numeric_col,"Array(Nullable(Decimal(38, 9)))" - array_bignumeric_col,"Array(Nullable(Decimal(76, 38)))" - array_bool_col,Array(Nullable(Bool)) - array_string_col,Array(Nullable(String)) - array_bytes_col,Array(Nullable(String)) - array_date_col,Array(Nullable(Date32)) - array_datetime_col,Array(Nullable(DateTime64(6))) - array_time_col,Array(Nullable(DateTime64(6))) - array_timestamp_col,Array(Nullable(DateTime64(6))) + datetime_col,"Nullable(DateTime64(6, 'UTC'))" + time_col,"Nullable(DateTime64(6, 'UTC'))" + timestamp_col,"Nullable(DateTime64(6, 'UTC'))" + array_int_col,Array(Int64) + array_float_col,Array(Float64) + array_numeric_col,"Array(Decimal(38, 9))" + array_bignumeric_col,"Array(Decimal(76, 38))" + array_bool_col,Array(Bool) + array_string_col,Array(String) + array_bytes_col,Array(String) + array_date_col,Array(Date32) + array_datetime_col,"Array(DateTime64(6, 'UTC'))" + array_time_col,"Array(DateTime64(6, 'UTC'))" + array_timestamp_col,"Array(DateTime64(6, 'UTC'))" - name: test_all_results_for_model_bigquery_to_duckdb_for_external_google_sheet resolver: sql properties: diff --git a/runtime/resolvers/testdata/connector_gcs.yaml b/runtime/resolvers/testdata/connector_gcs.yaml index 46966302ecba..83a33e82531c 100644 --- a/runtime/resolvers/testdata/connector_gcs.yaml +++ b/runtime/resolvers/testdata/connector_gcs.yaml @@ -106,7 +106,7 @@ tests: - name: test_all_results_for_model_gcs_to_clickhouse_for_glob resolver: sql properties: - sql: "select * from model_gcs_to_clickhouse_for_glob order by id" + sql: "select id, boolean_col, int32_col, int64_col, float_col, double_col, byte_array_col, fixed_len_byte_array_col, string_col, decimal_col, date_col, time_millis_col, time_micros_col, timestamp_millis_col, timestamp_micros_col, uuid_col, list_int_col, list_string_col, map_col, struct_col from model_gcs_to_clickhouse_for_glob order by id" connector: clickhouse # time is converted to datetime like 2025-01-01T00:02:03.456Z but should be 1970-01-01T00:02:03.456Z, which is wrong. result_csv: | diff --git a/runtime/resolvers/testdata/connector_mysql.yaml b/runtime/resolvers/testdata/connector_mysql.yaml index c71d6d1473c5..43a2d1733a56 100644 --- a/runtime/resolvers/testdata/connector_mysql.yaml +++ b/runtime/resolvers/testdata/connector_mysql.yaml @@ -67,7 +67,7 @@ tests: properties: sql: "select * from all_datatypes_clickhouse order by tinyint_col" connector: clickhouse - result_csv: "tinyint_col,smallint_col,mediumint_col,int_col,bigint_col,float_col,double_col,decimal_col,char_col,varchar_col,tinytext_col,text_col,mediumtext_col,longtext_col,binary_col,varbinary_col,tinyblob_col,blob_col,mediumblob_col,longblob_col,enum_col,set_col,date_col,datetime_col,timestamp_col,time_col,year_col,boolean_col,bit_col,json_col\n0,0,0,0,0,0,0,0.00,,,,,,,\0\0\0\0\0\0\0\0\0\0,,,,,,small,,1970-01-01,1970-01-01T00:00:00Z,,00:00:00,1970,0,0,{}\n127,32767,8388607,2147483647,1000000000000000000,1.100000023841858,2.2,3.30,C,VarChar,Tiny Text,Text,Medium Text,Long text content,Binary\0\0\0\0,VarBinary,Tiny Blob,Blob,Medium Blob,Long Blob,medium,\"a,b\",2024-02-14,2025-02-14T12:34:56Z,2025-02-14T12:34:56Z,12:34:56,2024,1,1,\"{\"\"key\"\": \"\"value\"\"}\"\n,,,,,,,,,,,,,,,,,,,,,,,,,,,,,\n" + result_csv: "tinyint_col,smallint_col,mediumint_col,int_col,bigint_col,float_col,double_col,decimal_col,char_col,varchar_col,tinytext_col,text_col,mediumtext_col,longtext_col,binary_col,varbinary_col,tinyblob_col,blob_col,mediumblob_col,longblob_col,enum_col,set_col,date_col,datetime_col,timestamp_col,time_col,year_col,boolean_col,bit_col,json_col\n0,0,0,0,0,0,0,0,,,,,,,\0\0\0\0\0\0\0\0\0\0,,,,,,small,,1970-01-01,1970-01-01T00:00:00Z,,00:00:00,1970,0,0,{}\n127,32767,8388607,2147483647,1000000000000000000,1.100000023841858,2.2,3.3,C,VarChar,Tiny Text,Text,Medium Text,Long text content,Binary\0\0\0\0,VarBinary,Tiny Blob,Blob,Medium Blob,Long Blob,medium,\"a,b\",2024-02-14,2025-02-14T12:34:56Z,2025-02-14T12:34:56Z,12:34:56,2024,1,1,\"{\"\"key\"\": \"\"value\"\"}\"\n,,,,,,,,,,,,,,,,,,,,,,,,,,,,,\n" - name: query_all_datatypes_clickhouse resolver: sql properties: @@ -83,7 +83,7 @@ tests: bigint_col,Nullable(Int64) float_col,Nullable(Float32) double_col,Nullable(Float64) - decimal_col,Nullable(String) + decimal_col,"Nullable(Decimal(10, 2))" char_col,Nullable(String) varchar_col,Nullable(String) tinytext_col,Nullable(String) @@ -98,8 +98,8 @@ tests: longblob_col,Nullable(String) enum_col,Nullable(String) set_col,Nullable(String) - date_col,Nullable(Date) - datetime_col,Nullable(DateTime) + date_col,Nullable(Date32) + datetime_col,Nullable(DateTime64(0)) timestamp_col,Nullable(DateTime) time_col,Nullable(String) year_col,Nullable(String) diff --git a/runtime/resolvers/testdata/connector_postgres.yaml b/runtime/resolvers/testdata/connector_postgres.yaml index 55aff4b8a1c1..19cb2e277a62 100644 --- a/runtime/resolvers/testdata/connector_postgres.yaml +++ b/runtime/resolvers/testdata/connector_postgres.yaml @@ -85,7 +85,7 @@ tests: name,Nullable(String) age,Nullable(Int32) is_married,Nullable(UInt8) - date_of_birth,Nullable(Date) + date_of_birth,Nullable(Date32) time_of_day,Nullable(String) created_at,Nullable(DateTime64(6)) personal_info,Nullable(String) diff --git a/runtime/resolvers/testdata/connector_s3.yaml b/runtime/resolvers/testdata/connector_s3.yaml index ae0c9e52a2dd..62d69d21126f 100644 --- a/runtime/resolvers/testdata/connector_s3.yaml +++ b/runtime/resolvers/testdata/connector_s3.yaml @@ -112,7 +112,7 @@ tests: resolver: sql properties: connector: clickhouse - sql: "select * from clickhouse_all_datatypes_glob order by id" + sql: "select id, boolean_col, int32_col, int64_col, float_col, double_col, byte_array_col, fixed_len_byte_array_col, string_col, decimal_col, date_col, time_millis_col, time_micros_col, timestamp_millis_col, timestamp_micros_col, uuid_col, list_int_col, list_string_col, map_col, struct_col from clickhouse_all_datatypes_glob order by id" # time is converted to datetime like 2025-01-01T00:02:03.456Z but should be 1970-01-01T00:02:03.456Z, which is wrong. result_csv: | id,boolean_col,int32_col,int64_col,float_col,double_col,byte_array_col,fixed_len_byte_array_col,string_col,decimal_col,date_col,time_millis_col,time_micros_col,timestamp_millis_col,timestamp_micros_col,uuid_col,list_int_col,list_string_col,map_col,struct_col @@ -151,10 +151,10 @@ tests: string_col,Nullable(String) decimal_col,"Nullable(Decimal(10, 2))" date_col,Nullable(Date32) - time_millis_col,Nullable(DateTime64(3)) - time_micros_col,Nullable(DateTime64(6)) - timestamp_millis_col,Nullable(DateTime64(3)) - timestamp_micros_col,Nullable(DateTime64(6)) + time_millis_col,"Nullable(DateTime64(3, 'UTC'))" + time_micros_col,"Nullable(DateTime64(6, 'UTC'))" + timestamp_millis_col,"Nullable(DateTime64(3, 'UTC'))" + timestamp_micros_col,"Nullable(DateTime64(6, 'UTC'))" uuid_col,Nullable(FixedString(16)) list_int_col,Array(Nullable(Int32)) list_string_col,Array(Nullable(String)) From d093f44b64c76406715b4e7e80d0aa956226b807 Mon Sep 17 00:00:00 2001 From: Anshul Khandelwal <12948312+k-anshul@users.noreply.github.com> Date: Mon, 20 Jul 2026 18:35:25 +0530 Subject: [PATCH 3/5] review --- runtime/drivers/clickhouse/crud.go | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/runtime/drivers/clickhouse/crud.go b/runtime/drivers/clickhouse/crud.go index 1d830aa28eb6..8fb1573dd742 100644 --- a/runtime/drivers/clickhouse/crud.go +++ b/runtime/drivers/clickhouse/crud.go @@ -620,16 +620,19 @@ func (c *Connection) getTableEngine(ctx context.Context, name string) (string, e return "", err } defer res.Close() - if !res.Next() { - if err := res.Err(); err != nil { + for res.Next() { + if err := res.Scan(&engine); err != nil { return "", err } - return "", fmt.Errorf("clickhouse: table %q not found", name) } - if err := res.Scan(&engine); err != nil { + err = res.Err() + if err != nil { return "", err } - return engine, res.Err() + if engine == "" { + return "", fmt.Errorf("clickhouse: table %q not found", name) + } + return engine, nil } func (c *Connection) getTablePartitions(ctx context.Context, name string) ([]string, error) { From 19bc2c6b8848922f0216096a09a6a399855c8875 Mon Sep 17 00:00:00 2001 From: Anshul Khandelwal <12948312+k-anshul@users.noreply.github.com> Date: Mon, 20 Jul 2026 19:01:04 +0530 Subject: [PATCH 4/5] nit --- runtime/drivers/clickhouse/crud.go | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/runtime/drivers/clickhouse/crud.go b/runtime/drivers/clickhouse/crud.go index 8fb1573dd742..f55dee24bf13 100644 --- a/runtime/drivers/clickhouse/crud.go +++ b/runtime/drivers/clickhouse/crud.go @@ -3,6 +3,7 @@ package clickhouse import ( "context" "crypto/md5" + "database/sql" "errors" "fmt" "strings" @@ -629,9 +630,6 @@ func (c *Connection) getTableEngine(ctx context.Context, name string) (string, e if err != nil { return "", err } - if engine == "" { - return "", fmt.Errorf("clickhouse: table %q not found", name) - } return engine, nil } From 7eccde10de1ed84e2aadb5ac3e91ae2c0c6ecda0 Mon Sep 17 00:00:00 2001 From: Anshul Khandelwal <12948312+k-anshul@users.noreply.github.com> Date: Tue, 21 Jul 2026 09:24:46 +0530 Subject: [PATCH 5/5] build fix --- runtime/drivers/clickhouse/crud.go | 1 - 1 file changed, 1 deletion(-) diff --git a/runtime/drivers/clickhouse/crud.go b/runtime/drivers/clickhouse/crud.go index f55dee24bf13..c54e7bba4ee0 100644 --- a/runtime/drivers/clickhouse/crud.go +++ b/runtime/drivers/clickhouse/crud.go @@ -3,7 +3,6 @@ package clickhouse import ( "context" "crypto/md5" - "database/sql" "errors" "fmt" "strings"