Adding experimental (and limited) support for table partitioning

gitignore

cont'd

# Conflicts:
#	.gitignore
#	platform/ingest/processor.go

Author: mieciu

platform/clickhouse/clickhouse.go

@@ -26,6 +26,17 @@ const (
 	allElasticsearchIndicesPattern = "_all"
 )
 
+// PartitionStrategy is a custom type for partitioning strategies
+type PartitionStrategy string
+
+// Enum values for PartitionStrategy
+const (
+	Hourly  PartitionStrategy = "hourly"
+	Daily   PartitionStrategy = "daily"
+	Monthly PartitionStrategy = "monthly"
+	Yearly  PartitionStrategy = "yearly"
+)
+
 type (
 	LogManager struct {
 		ctx            context.Context
@@ -55,10 +66,11 @@ type (
 		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
-		PrimaryKey           string // "" if none
-		Settings             string // "" if none
-		Ttl                  string // of type Interval, e.g. 3 MONTH, 1 YEAR
+		//PartitionBy          string // "" if none
+		PartitionStrategy PartitionStrategy
+		PrimaryKey        string // "" if none
+		Settings          string // "" if none
+		Ttl               string // of type Interval, e.g. 3 MONTH, 1 YEAR
 		// look https://clickhouse.com/docs/en/sql-reference/data-types/special-data-types/interval
 		// "" if none
 		// TODO make sure it's unique in schema (there's no other 'others' field)
@@ -360,9 +372,10 @@ func NewDefaultCHConfig() *ChTableConfig {
 		TimestampDefaultsNow: true,
 		Engine:               "MergeTree",
 		OrderBy:              "(" + `"@timestamp"` + ")",
-		PartitionBy:          "",
-		PrimaryKey:           "",
-		Ttl:                  "",
+		//PartitionBy:          "",
+		PartitionStrategy: Hourly,
+		PrimaryKey:        "",
+		Ttl:               "",
 		Attributes: []Attribute{
 			NewDefaultInt64Attribute(),
 			NewDefaultFloat64Attribute(),
@@ -381,9 +394,9 @@ func NewNoTimestampOnlyStringAttrCHConfig() *ChTableConfig {
 		TimestampDefaultsNow: false,
 		Engine:               "MergeTree",
 		OrderBy:              "(" + `"@timestamp"` + ")",
-		PartitionBy:          "",
-		PrimaryKey:           "",
-		Ttl:                  "",
+		//PartitionBy:          "",
+		PrimaryKey: "",
+		Ttl:        "",
 		Attributes: []Attribute{
 			NewDefaultStringAttribute(),
 		},

platform/clickhouse/schema.go

@@ -8,6 +8,7 @@ import (
 	"github.com/QuesmaOrg/quesma/platform/util"
 	"math"
 	"reflect"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -342,8 +343,8 @@ func (config *ChTableConfig) CreateTablePostFieldsString() string {
 	if config.OrderBy != "" {
 		s += "ORDER BY " + config.OrderBy + "\n"
 	}
-	if config.PartitionBy != "" {
-		s += "PARTITION BY " + config.PartitionBy + "\n"
+	if config.PartitionStrategy != "" {
+		s += "PARTITION BY " + getPartitioningFunc(config.PartitionStrategy) + "(" + strconv.Quote(timestampFieldName) + ")" + "\n"
 	}
 	if config.PrimaryKey != "" {
 		s += "PRIMARY KEY " + config.PrimaryKey + "\n"
@@ -358,6 +359,21 @@ func (config *ChTableConfig) CreateTablePostFieldsString() string {
 	return s
 }
 
+func getPartitioningFunc(strategy PartitionStrategy) string {
+	switch strategy {
+	case Hourly:
+		return "toStartOfHour"
+	case Daily:
+		return "toYYYYMMDD"
+	case Monthly:
+		return "toYYYYMM"
+	case Yearly:
+		return "toYYYY"
+	default:
+		return ""
+	}
+}
+
 func NewDefaultStringAttribute() Attribute {
 	return Attribute{
 		KeysArrayName:   DeprecatedAttributesKeyColumn,

platform/clickhouse/table.go

@@ -73,6 +73,7 @@ func (t *Table) CreateTableString() string {
 	for _, index := range t.Indexes {
 		rows = append(rows, util.Indent(1)+index.Statement())
 	}
+	t.Config.PartitionStrategy = Hourly
 	return s + strings.Join(rows, ",\n") + "\n)\n" + t.Config.CreateTablePostFieldsString()
 }
 

platform/ingest/processor.go

@@ -229,6 +229,7 @@ func createTableQuery(name string, columns string, config *chLib.ChTableConfig)
 	if config.ClusterName != "" {
 		onClusterClause = "ON CLUSTER " + strconv.Quote(config.ClusterName) + " "
 	}
+	config.PartitionStrategy = chLib.Hourly
 	createTableCmd := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS "%s" %s
 (
 
@@ -1008,11 +1009,11 @@ func NewIngestProcessor(cfg *config.QuesmaConfiguration, chDb quesma_api.Backend
 
 func NewOnlySchemaFieldsCHConfig(clusterName string) *chLib.ChTableConfig {
 	return &chLib.ChTableConfig{
-		HasTimestamp:                          true,
-		TimestampDefaultsNow:                  true,
-		Engine:                                "MergeTree",
-		OrderBy:                               "(" + `"@timestamp"` + ")",
-		PartitionBy:                           "",
+		HasTimestamp:         true,
+		TimestampDefaultsNow: true,
+		Engine:               "MergeTree",
+		OrderBy:              "(" + `"@timestamp"` + ")",
+		//PartitionBy:                           "",
 		ClusterName:                           clusterName,
 		PrimaryKey:                            "",
 		Ttl:                                   "",
@@ -1029,9 +1030,9 @@ func NewDefaultCHConfig() *chLib.ChTableConfig {
 		TimestampDefaultsNow: true,
 		Engine:               "MergeTree",
 		OrderBy:              "(" + `"@timestamp"` + ")",
-		PartitionBy:          "",
-		PrimaryKey:           "",
-		Ttl:                  "",
+		//PartitionBy:          "",
+		PrimaryKey: "",
+		Ttl:        "",
 		Attributes: []chLib.Attribute{
 			chLib.NewDefaultInt64Attribute(),
 			chLib.NewDefaultFloat64Attribute(),