Adding support for ON CLUSTER clause in DDL statements (#1267)

This PR adds support for `ON CLUSTER` clause within DDL statements. If
user want to synchronize DDL across nodes, it is required to place
`clusterName` in the ingest processor configuration like this:
```yaml
  - name: my-ingest-processor
    type: quesma-v1-processor-ingest
    config:
      clusterName: "quesma_cluster"
      indexes:
        example:
          target: [ my-clickhouse-data-source ]
```

Having that, Quesma is going to append `ON CLUSTER "quesma_cluster"` to
all `CREATE TABLE` and `ALTER TABLE` statements.


###  Some useful bits for testing

**Step 0.** Make sure you configured `clusterName` for table named
`try_it`.
  
Test table creation:
```
curl -i -X POST "http://localhost:8080/_bulk" -H "Content-Type: application/ndjson" -d '
{ "index": { "_index": "try_it", "_id": "1" } }
{ "name": "Alice", "age": 30, "city": "New York" }
{ "index": { "_index": "try_it", "_id": "2" } }
{ "name": "Bob", "age": 25, "city": "San Francisco" }
{ "index": { "_index": "try_it", "_id": "3" } }
{ "name": "Charlie", "age": 35, "city": "Los Angeles" }
'
```
Test `ALTER TABLE`:
```
curl -i -X POST "http://localhost:8080/_bulk" -H "Content-Type: application/ndjson" -d '
{ "index": { "_index": "try_it", "_id": "6" } }
{ "name": "Alice", "age": 30, "city": "New York", "genre": "jazz" }
{ "index": { "_index": "try_it", "_id": "8" } }
{ "name": "Bob", "age": 25, "city": "San Francisco", "genre": "hiphop" }
'
```

Also the same steps (create + alter) should be run for common table
scenario.

---------

Co-authored-by: Piotr Grabowski <[email protected]>

Author: mieciu

docs/public/docs/config-primer.md

@@ -106,6 +106,7 @@ The supported configuration options for backend connectors (under `config`):
 * `user` - username for authentication
 * `password` - password for authentication 
 * `database` - name of the database to connect to. It is optional for ClickHouse, but strictly required for Hydrolix, where it is also referred as "project".
+* `clusterName` - name of the ClickHouse cluster (optional). This will be used in the `ON CLUSTER clusterName` clause when Quesma creates tables. Setting this option ensures that the created tables are present on all nodes of the distributed cluster.
 * `adminUrl` - URL for administrative operations to render a handy link in Quesma management UI (optional)
 * `disableTLS` - when set to true, disables TLS for the connection (optional)
 

docs/public/docs/limitations.md

@@ -24,6 +24,7 @@ Quesma supports only ElasticSearch/Kibana 8.0 or above.
 
 ### ClickHouse limitations
 * When using a cluster deployment of ClickHouse, the tables automatically created by Quesma (during [Ingest](/ingest.md)) will use the `MergeTree` engine. If you wish to use the `ReplicatedMergeTree` engine instead, you will have to create the tables manually with  `ReplicatedMergeTree` engine before ingesting data to Quesma.
+  * Remember to configure `clusterName` in backend connector configuration to make sure that the tables are created on all nodes of the cluster. 
   * *Note: On ClickHouse Cloud, the tables automatically created by Quesma will use the `ReplicatedMergeTree` engine (ClickHouse Cloud default engine).* 
 
 ## Functional limitations

quesma/clickhouse/clickhouse.go

@@ -52,6 +52,7 @@ type (
 		// TODO make sure it's unique in schema (there's no other 'timestamp' field)
 		// I (Krzysiek) can write it quickly, but don't want to waste time for it right now.
 		TimestampDefaultsNow bool
+		ClusterName          string // Name of the cluster if created with `CREATE TABLE ... ON CLUSTER ClusterName`
 		Engine               string // "Log", "MergeTree", etc.
 		OrderBy              string // "" if none
 		PartitionBy          string // "" if none
@@ -379,6 +380,7 @@ func NewDefaultCHConfig() *ChTableConfig {
 	}
 }
 
+// NewNoTimestampOnlyStringAttrCHConfig is used only in tests
 func NewNoTimestampOnlyStringAttrCHConfig() *ChTableConfig {
 	return &ChTableConfig{
 		HasTimestamp:         false,
@@ -396,6 +398,7 @@ func NewNoTimestampOnlyStringAttrCHConfig() *ChTableConfig {
 	}
 }
 
+// NewChTableConfigNoAttrs is used only in tests
 func NewChTableConfigNoAttrs() *ChTableConfig {
 	return &ChTableConfig{
 		HasTimestamp:                          false,
@@ -408,6 +411,7 @@ func NewChTableConfigNoAttrs() *ChTableConfig {
 	}
 }
 
+// NewChTableConfigTimestampStringAttr is used only in tests
 func NewChTableConfigTimestampStringAttr() *ChTableConfig {
 	return &ChTableConfig{
 		HasTimestamp:                          true,

quesma/clickhouse/parserCreateTable.go

@@ -389,11 +389,22 @@ func ParseCreateTable(q string) (*Table, int) {
 	// parse [ON CLUSTER cluster_name]
 	i3 := parseExact(q, i2, "ON CLUSTER ")
 	if i3 != -1 {
+		i3 = omitWhitespace(q, i3)
+		i4, _ := parseMaybeAndForget(q, i3, `"`) // cluster name can be quoted, but doesn't have to
+		if i4 != -1 {
+			i3 = i4
+		}
 		i4, ident := parseIdent(q, i3)
 		if i4 == -1 {
 			return &t, i3
 		}
-		t.Cluster = ident
+		t.ClusterName = ident
+		if i4 != -1 {
+			i4, _ = parseMaybeAndForget(q, i4, `"`)
+			if i4 == -1 {
+				return &t, i3
+			}
+		}
 		i2 = i4
 	}
 

quesma/clickhouse/schema.go

@@ -314,6 +314,7 @@ func NewTable(createTableQuery string, config *ChTableConfig) (*Table, error) {
 	}
 }
 
+// NewEmptyTable is used only in tests
 func NewEmptyTable(tableName string) *Table {
 	return &Table{Name: tableName, Config: NewChTableConfigNoAttrs()}
 }

quesma/clickhouse/table.go

@@ -16,7 +16,7 @@ import (
 type Table struct {
 	Name         string
 	DatabaseName string `default:""`
-	Cluster      string `default:""`
+	ClusterName  string `default:""`
 	Cols         map[string]*Column
 	Config       *ChTableConfig
 	Created      bool // do we need to create it during first insert
@@ -60,7 +60,11 @@ func (t *Table) createTableOurFieldsString() []string {
 }
 
 func (t *Table) CreateTableString() string {
-	s := "CREATE TABLE IF NOT EXISTS " + t.FullTableName() + " (\n"
+	var onClusterClause string
+	if t.ClusterName != "" {
+		onClusterClause = " ON CLUSTER " + strconv.Quote(t.ClusterName)
+	}
+	s := "CREATE TABLE IF NOT EXISTS " + t.FullTableName() + onClusterClause + " (\n"
 	rows := make([]string, 0)
 	for _, col := range t.Cols {
 		rows = append(rows, col.createTableString(1))

quesma/clickhouse/table_discovery.go

@@ -398,6 +398,7 @@ func (td *tableDiscovery) populateTableDefinitions(configuredTables map[string]d
 				Name:         tableName,
 				Comment:      resTable.comment,
 				DatabaseName: databaseName,
+				ClusterName:  cfg.ClusterName, // FIXME: is this really necessary? The cluster name is only used when creating table, but this is an already created table - so is this information not needed?
 				Cols:         columnsMap,
 				Config: &ChTableConfig{
 					Attributes:                            []Attribute{},

quesma/common_table/const.go

@@ -4,15 +4,17 @@ package common_table
 
 import (
 	"context"
+	"fmt"
 	"github.com/QuesmaOrg/quesma/quesma/logger"
 	quesma_api "github.com/QuesmaOrg/quesma/quesma/v2/core"
+	"strconv"
 )
 
 const TableName = "quesma_common_table"
 const IndexNameColumn = "__quesma_index_name"
 
 const singleTableDDL = `
-CREATE TABLE IF NOT EXISTS "quesma_common_table"
+CREATE TABLE IF NOT EXISTS "quesma_common_table" %s
 (
     "attributes_values" Map(String, String),
     "attributes_metadata" Map(String, String),
@@ -26,10 +28,19 @@ CREATE TABLE IF NOT EXISTS "quesma_common_table"
 
 `
 
-func EnsureCommonTableExists(db quesma_api.BackendConnector) {
+func commonTableDDL(clusterName string) string {
+	var maybeOnClusterClause string
+	if clusterName != "" {
+		maybeOnClusterClause = "ON CLUSTER " + strconv.Quote(clusterName)
+	}
+	return fmt.Sprintf(singleTableDDL, maybeOnClusterClause)
+
+}
+
+func EnsureCommonTableExists(db quesma_api.BackendConnector, clusterName string) {
 
 	logger.Info().Msgf("Ensuring common table '%v' exists", TableName)
-	err := db.Exec(context.Background(), singleTableDDL)
+	err := db.Exec(context.Background(), commonTableDDL(clusterName))
 	if err != nil {
 		// TODO check if we've got RO access to the database
 		logger.Warn().Msgf("Failed to create common table '%v': %v", TableName, err)

quesma/ingest/parserCreateTable_test.go

@@ -77,7 +77,7 @@ func TestParseSignozSchema_2(t *testing.T) {
 		assert.Contains(t, table.Cols, fieldName)
 	}
 	assert.Equal(t, "db", table.DatabaseName)
-	assert.Equal(t, "cluster", table.Cluster)
+	assert.Equal(t, "cluster", table.ClusterName)
 }
 
 func TestParseQuotedTablename(t *testing.T) {
@@ -119,6 +119,79 @@ func TestParseNonLetterNames(t *testing.T) {
 	}
 }
 
+func TestParseCreateSampleDataEcommerce(t *testing.T) {
+	q := `CREATE TABLE IF NOT EXISTS "kibana_sample_data_ecommerce" ON CLUSTER "quesma_cluster" 
+(
+	"@timestamp" DateTime64(3) DEFAULT now64(),
+	"attributes_values" Map(String,String),
+	"attributes_metadata" Map(String,String),
+
+
+	"taxful_total_price" Nullable(Float64) COMMENT  'quesmaMetadataV1:fieldName=taxful_total_price',
+	"sku" Array(String) COMMENT  'quesmaMetadataV1:fieldName=sku',
+	"taxless_total_price" Nullable(Float64) COMMENT  'quesmaMetadataV1:fieldName=taxless_total_price',
+	"total_unique_products" Nullable(Int64) COMMENT  'quesmaMetadataV1:fieldName=total_unique_products',
+	"geoip_region_name" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=geoip.region_name',
+	"category" Array(String) COMMENT  'quesmaMetadataV1:fieldName=category',
+	"products_created_on" Array(DateTime64) COMMENT  'quesmaMetadataV1:fieldName=products.created_on',
+	"products_taxful_price" Array(Float64) COMMENT  'quesmaMetadataV1:fieldName=products.taxful_price',
+	"user" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=user',
+	"currency" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=currency',
+	"day_of_week" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=day_of_week',
+	"geoip_location_lat" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=geoip.location.lat',
+	"geoip_city_name" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=geoip.city_name',
+	"products__id" Array(String) COMMENT  'quesmaMetadataV1:fieldName=products._id',
+	"products_discount_amount" Array(Int64) COMMENT  'quesmaMetadataV1:fieldName=products.discount_amount',
+	"customer_last_name" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=customer_last_name',
+	"total_quantity" Nullable(Int64) COMMENT  'quesmaMetadataV1:fieldName=total_quantity',
+	"geoip_continent_name" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=geoip.continent_name',
+	"products_price" Array(Float64) COMMENT  'quesmaMetadataV1:fieldName=products.price',
+	"products_sku" Array(String) COMMENT  'quesmaMetadataV1:fieldName=products.sku',
+	"products_discount_percentage" Array(Int64) COMMENT  'quesmaMetadataV1:fieldName=products.discount_percentage',
+	"type" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=type',
+	"customer_full_name" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=customer_full_name',
+	"products_min_price" Array(Float64) COMMENT  'quesmaMetadataV1:fieldName=products.min_price',
+	"products_unit_discount_amount" Array(Int64) COMMENT  'quesmaMetadataV1:fieldName=products.unit_discount_amount',
+	"products_manufacturer" Array(String) COMMENT  'quesmaMetadataV1:fieldName=products.manufacturer',
+	"products_base_unit_price" Array(Float64) COMMENT  'quesmaMetadataV1:fieldName=products.base_unit_price',
+	"day_of_week_i" Nullable(Int64) COMMENT  'quesmaMetadataV1:fieldName=day_of_week_i',
+	"manufacturer" Array(String) COMMENT  'quesmaMetadataV1:fieldName=manufacturer',
+	"customer_first_name" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=customer_first_name',
+	"products_product_id" Array(Int64) COMMENT  'quesmaMetadataV1:fieldName=products.product_id',
+	"customer_gender" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=customer_gender',
+	"email" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=email',
+	"order_id" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=order_id',
+	"customer_phone" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=customer_phone',
+	"order_date" Nullable(DateTime64) COMMENT  'quesmaMetadataV1:fieldName=order_date',
+	"geoip_location_lon" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=geoip.location.lon',
+	"products_base_price" Array(Float64) COMMENT  'quesmaMetadataV1:fieldName=products.base_price',
+	"products_category" Array(String) COMMENT  'quesmaMetadataV1:fieldName=products.category',
+	"products_tax_amount" Array(Int64) COMMENT  'quesmaMetadataV1:fieldName=products.tax_amount',
+	"products_product_name" Array(String) COMMENT  'quesmaMetadataV1:fieldName=products.product_name',
+	"customer_id" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=customer_id',
+	"event_dataset" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=event.dataset',
+	"geoip_country_iso_code" Nullable(String) COMMENT  'quesmaMetadataV1:fieldName=geoip.country_iso_code',
+	"products_taxless_price" Array(Float64) COMMENT  'quesmaMetadataV1:fieldName=products.taxless_price',
+	"products_quantity" Array(Int64) COMMENT  'quesmaMetadataV1:fieldName=products.quantity',
+	"customer_birth_date" Nullable(DateTime64) COMMENT  'quesmaMetadataV1:fieldName=customer_birth_date',
+
+)
+ENGINE = MergeTree
+ORDER BY ("@timestamp")
+
+COMMENT 'created by Quesma'`
+
+	fieldNames := []string{"@timestamp", "attributes_values", "attributes_metadata", "taxful_total_price", "sku", "taxless_total_price", "total_unique_products", "geoip_region_name", "category", "products_created_on", "products_taxful_price", "user", "currency", "day_of_week", "geoip_location_lat", "geoip_city_name", "products__id", "products_discount_amount", "customer_last_name", "total_quantity", "geoip_continent_name", "products_price", "products_sku", "products_discount_percentage", "type", "customer_full_name", "products_min_price", "products_unit_discount_amount", "products_manufacturer", "products_base_unit_price", "day_of_week_i", "manufacturer", "customer_first_name", "products_product_id", "customer_gender", "email", "order_id", "customer_phone", "order_date", "geoip_location_lon", "products_base_price", "products_category", "products_tax_amount", "products_product_name", "customer_id", "event_dataset", "geoip_country_iso_code", "products_taxless_price", "products_quantity", "customer_birth_date"}
+	table, err := clickhouse.NewTable(q, nil)
+	assert.NoError(t, err)
+	assert.Equal(t, len(fieldNames), len(table.Cols))
+	for _, fieldName := range fieldNames {
+		assert.Contains(t, table.Cols, fieldName)
+	}
+	assert.Equal(t, "kibana_sample_data_ecommerce", table.Name)
+	assert.Equal(t, "quesma_cluster", table.ClusterName)
+}
+
 func TestParseLongNestedSchema(t *testing.T) {
 	q := `CREATE TABLE IF NOT EXISTS "/_monitoring/bulk?system_id=kibana&system_api_version=7&interval=10000ms_2"
 		(

quesma/ingest/processor.go

@@ -26,6 +26,7 @@ import (
 	"github.com/goccy/go-json"
 	"slices"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -224,14 +225,18 @@ func Indexes(m SchemaMap) string {
 }
 
 func createTableQuery(name string, columns string, config *chLib.ChTableConfig) string {
-	createTableCmd := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s"
+	var onClusterClause string
+	if config.ClusterName != "" {
+		onClusterClause = "ON CLUSTER " + strconv.Quote(config.ClusterName) + " "
+	}
+	createTableCmd := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s" %s
 (
 
 %s
 )
 %s
 COMMENT 'created by Quesma'`,
-		name, columns,
+		name, onClusterClause, columns,
 		config.CreateTablePostFieldsString())
 	return createTableCmd
 }
@@ -337,11 +342,16 @@ func (ip *IngestProcessor) generateNewColumns(
 		metadata.Values[comment_metadata.ElasticFieldName] = propertyName
 		comment := metadata.Marshall()
 
-		alterTable := fmt.Sprintf("ALTER TABLE \"%s\" ADD COLUMN IF NOT EXISTS \"%s\" %s", table.Name, attrKeys[i], columnType)
+		var maybeOnClusterClause string
+		if table.ClusterName != "" {
+			maybeOnClusterClause = " ON CLUSTER " + strconv.Quote(table.ClusterName)
+		}
+
+		alterTable := fmt.Sprintf("ALTER TABLE \"%s\"%s ADD COLUMN IF NOT EXISTS \"%s\" %s", table.Name, maybeOnClusterClause, attrKeys[i], columnType)
 		newColumns[attrKeys[i]] = &chLib.Column{Name: attrKeys[i], Type: chLib.NewBaseType(attrTypes[i]), Modifiers: modifiers, Comment: comment}
 		alterCmd = append(alterCmd, alterTable)
 
-		alterColumn := fmt.Sprintf("ALTER TABLE \"%s\" COMMENT COLUMN \"%s\" '%s'", table.Name, attrKeys[i], comment)
+		alterColumn := fmt.Sprintf("ALTER TABLE \"%s\"%s COMMENT COLUMN \"%s\" '%s'", table.Name, maybeOnClusterClause, attrKeys[i], comment)
 		alterCmd = append(alterCmd, alterColumn)
 
 		deleteIndexes = append(deleteIndexes, i)
@@ -632,7 +642,7 @@ func (ip *IngestProcessor) processInsertQuery(ctx context.Context,
 	var tableConfig *chLib.ChTableConfig
 	var createTableCmd string
 	if table == nil {
-		tableConfig = NewOnlySchemaFieldsCHConfig()
+		tableConfig = NewOnlySchemaFieldsCHConfig(ip.cfg.ClusterName)
 		columnsFromJson := JsonToColumns(transformedJsons[0], tableConfig)
 
 		fieldOrigins := make(map[schema.FieldName]schema.FieldSource)
@@ -996,13 +1006,14 @@ func NewIngestProcessor(cfg *config.QuesmaConfiguration, chDb quesma_api.Backend
 	return &IngestProcessor{ctx: ctx, cancel: cancel, chDb: chDb, tableDiscovery: loader, cfg: cfg, phoneHomeClient: phoneHomeClient, schemaRegistry: schemaRegistry, virtualTableStorage: virtualTableStorage, tableResolver: tableResolver}
 }
 
-func NewOnlySchemaFieldsCHConfig() *chLib.ChTableConfig {
+func NewOnlySchemaFieldsCHConfig(clusterName string) *chLib.ChTableConfig {
 	return &chLib.ChTableConfig{
 		HasTimestamp:                          true,
 		TimestampDefaultsNow:                  true,
 		Engine:                                "MergeTree",
 		OrderBy:                               "(" + `"@timestamp"` + ")",
 		PartitionBy:                           "",
+		ClusterName:                           clusterName,
 		PrimaryKey:                            "",
 		Ttl:                                   "",
 		Attributes:                            []chLib.Attribute{chLib.NewDefaultStringAttribute()},
@@ -1011,6 +1022,7 @@ func NewOnlySchemaFieldsCHConfig() *chLib.ChTableConfig {
 	}
 }
 
+// NewDefaultCHConfig is used only in tests
 func NewDefaultCHConfig() *chLib.ChTableConfig {
 	return &chLib.ChTableConfig{
 		HasTimestamp:         true,
@@ -1031,6 +1043,7 @@ func NewDefaultCHConfig() *chLib.ChTableConfig {
 	}
 }
 
+// NewChTableConfigNoAttrs is used only in tests
 func NewChTableConfigNoAttrs() *chLib.ChTableConfig {
 	return &chLib.ChTableConfig{
 		HasTimestamp:                          false,
@@ -1043,6 +1056,7 @@ func NewChTableConfigNoAttrs() *chLib.ChTableConfig {
 	}
 }
 
+// NewChTableConfigFourAttrs is used only in tests
 func NewChTableConfigFourAttrs() *chLib.ChTableConfig {
 	return &chLib.ChTableConfig{
 		HasTimestamp:         false,