Add support of parsing TTL policy when creating table (#137)

Author: git-hulk

parser/ast.go

@@ -7,7 +7,7 @@ import (
 type OrderDirection string
 
 const (
-	OrderDirectionNone OrderDirection = "None"
+	OrderDirectionNone OrderDirection = ""
 	OrderDirectionAsc  OrderDirection = "ASC"
 	OrderDirectionDesc OrderDirection = "DESC"
 )
@@ -2690,9 +2690,164 @@ func (s *SampleByClause) Accept(visitor ASTVisitor) error {
 	return visitor.VisitSampleByExpr(s)
 }
 
+type TTLPolicyRuleAction struct {
+	ActionPos Pos
+	ActionEnd Pos
+	Action    string
+	Codec     *CompressionCodec
+}
+
+func (t *TTLPolicyRuleAction) Pos() Pos {
+	return t.ActionPos
+}
+
+func (t *TTLPolicyRuleAction) End() Pos {
+	if t.Codec != nil {
+		return t.Codec.End()
+	}
+	return t.ActionEnd
+}
+
+func (t *TTLPolicyRuleAction) String() string {
+	var builder strings.Builder
+	builder.WriteString(t.Action)
+	if t.Codec != nil {
+		builder.WriteString(" ")
+		builder.WriteString(t.Codec.String())
+	}
+	return builder.String()
+}
+
+func (t *TTLPolicyRuleAction) Accept(visitor ASTVisitor) error {
+	visitor.enter(t)
+	defer visitor.leave(t)
+	if t.Codec != nil {
+		if err := t.Codec.Accept(visitor); err != nil {
+			return err
+		}
+	}
+	return visitor.VisitTTLPolicyItemAction(t)
+}
+
+type TTLPolicyRule struct {
+	RulePos  Pos
+	ToVolume *StringLiteral
+	ToDisk   *StringLiteral
+	Action   *TTLPolicyRuleAction
+}
+
+func (t *TTLPolicyRule) Pos() Pos {
+	return t.RulePos
+}
+
+func (t *TTLPolicyRule) End() Pos {
+	if t.Action != nil {
+		return t.Action.End()
+	}
+	if t.ToDisk != nil {
+		return t.ToDisk.LiteralEnd
+	}
+	return t.ToVolume.LiteralEnd
+}
+
+func (t *TTLPolicyRule) String() string {
+	var builder strings.Builder
+	if t.ToVolume != nil {
+		builder.WriteString("TO VOLUME ")
+		builder.WriteString(t.ToVolume.String())
+	} else if t.ToDisk != nil {
+		builder.WriteString("TO DISK ")
+		builder.WriteString(t.ToDisk.String())
+	} else if t.Action != nil {
+		builder.WriteString(t.Action.String())
+	}
+	return builder.String()
+}
+
+func (t *TTLPolicyRule) Accept(visitor ASTVisitor) error {
+	visitor.enter(t)
+	defer visitor.leave(t)
+	if t.ToVolume != nil {
+		if err := t.ToVolume.Accept(visitor); err != nil {
+			return err
+		}
+	}
+	if t.ToDisk != nil {
+		if err := t.ToDisk.Accept(visitor); err != nil {
+			return err
+		}
+	}
+	return visitor.VisitTTLPolicyRule(t)
+}
+
+type TTLPolicy struct {
+	Item    *TTLPolicyRule
+	Where   *WhereClause
+	GroupBy *GroupByClause
+}
+
+func (t *TTLPolicy) Pos() Pos {
+	if t.Item != nil {
+		return t.Item.Pos()
+	}
+	if t.Where != nil {
+		return t.Where.Pos()
+	}
+	return t.GroupBy.Pos()
+}
+
+func (t *TTLPolicy) End() Pos {
+	if t.GroupBy != nil {
+		return t.GroupBy.End()
+	}
+	if t.Where != nil {
+		return t.Where.End()
+	}
+	return t.Item.End()
+}
+
+func (t *TTLPolicy) String() string {
+	var builder strings.Builder
+
+	if t.Item != nil {
+		builder.WriteString(t.Item.String())
+	}
+	if t.Where != nil {
+		builder.WriteString(" ")
+		builder.WriteString(t.Where.String())
+	}
+	if t.GroupBy != nil {
+		builder.WriteString(" ")
+		builder.WriteString(t.GroupBy.String())
+	}
+	return builder.String()
+}
+
+func (t *TTLPolicy) Accept(visitor ASTVisitor) error {
+	visitor.enter(t)
+	defer visitor.leave(t)
+	if t.Item != nil {
+		if err := t.Item.Accept(visitor); err != nil {
+			return err
+		}
+	}
+	if t.Where != nil {
+		if err := t.Where.Accept(visitor); err != nil {
+			return err
+		}
+	}
+	if t.GroupBy != nil {
+		if err := t.GroupBy.Accept(visitor); err != nil {
+			return err
+		}
+	}
+	return visitor.VisitTTLPolicy(t)
+}
+
 type TTLExpr struct {
 	TTLPos Pos
 	Expr   Expr
+	Policy *TTLPolicy
 }
 
 func (t *TTLExpr) Pos() Pos {
@@ -2706,6 +2861,10 @@ func (t *TTLExpr) End() Pos {
 func (t *TTLExpr) String() string {
 	var builder strings.Builder
 	builder.WriteString(t.Expr.String())
+	if t.Policy != nil {
+		builder.WriteString(" ")
+		builder.WriteString(t.Policy.String())
+	}
 	return builder.String()
 }
 
@@ -2715,6 +2874,11 @@ func (t *TTLExpr) Accept(visitor ASTVisitor) error {
 	if err := t.Expr.Accept(visitor); err != nil {
 		return err
 	}
+	if t.Policy != nil {
+		if err := t.Policy.Accept(visitor); err != nil {
+			return err
+		}
+	}
 	return visitor.VisitTTLExpr(t)
 }
 

parser/ast_visitor.go

@@ -61,6 +61,9 @@ type ASTVisitor interface {
 	VisitSampleByExpr(expr *SampleByClause) error
 	VisitTTLExpr(expr *TTLExpr) error
 	VisitTTLExprList(expr *TTLClause) error
+	VisitTTLPolicy(expr *TTLPolicy) error
+	VisitTTLPolicyRule(expr *TTLPolicyRule) error
+	VisitTTLPolicyItemAction(expr *TTLPolicyRuleAction) error
 	VisitOrderByExpr(expr *OrderExpr) error
 	VisitOrderByListExpr(expr *OrderByClause) error
 	VisitSettingsExpr(expr *SettingExprList) error
@@ -596,6 +599,27 @@ func (v *DefaultASTVisitor) VisitTTLExprList(expr *TTLClause) error {
 	return nil
 }
 
+func (v *DefaultASTVisitor) VisitTTLPolicy(expr *TTLPolicy) error {
+	if v.Visit != nil {
+		return v.Visit(expr)
+	}
+	return nil
+}
+
+func (v *DefaultASTVisitor) VisitTTLPolicyRule(expr *TTLPolicyRule) error {
+	if v.Visit != nil {
+		return v.Visit(expr)
+	}
+	return nil
+}
+
+func (v *DefaultASTVisitor) VisitTTLPolicyItemAction(expr *TTLPolicyRuleAction) error {
+	if v.Visit != nil {
+		return v.Visit(expr)
+	}
+	return nil
+}
+
 func (v *DefaultASTVisitor) VisitOrderByExpr(expr *OrderExpr) error {
 	if v.Visit != nil {
 		return v.Visit(expr)

parser/keyword.go

@@ -158,6 +158,7 @@ const (
 	KeywordQueues       = "QUEUES"
 	KeywordQuota        = "QUOTA"
 	KeywordRange        = "RANGE"
+	KeywordRecompress   = "RECOMPRESS"
 	KeywordRefresh      = "REFRESH"
 	KeywordRegexp       = "REGEXP"
 	KeywordReload       = "RELOAD"
@@ -385,6 +386,7 @@ var keywords = NewSet(
 	KeywordQueues,
 	KeywordQuota,
 	KeywordRange,
+	KeywordRecompress,
 	KeywordRefresh,
 	KeywordRegexp,
 	KeywordReload,

parser/parser_table.go

@@ -691,6 +691,11 @@ func (p *Parser) parseOrderExpr(pos Pos) (*OrderExpr, error) {
 		if err != nil {
 			return nil, err
 		}
+	} else if p.matchKeyword(KeywordTtl) {
+		return &OrderExpr{
+			OrderPos: pos,
+			Expr:     columnExpr,
+		}, nil
 	}
 
 	direction := OrderDirectionNone
@@ -746,28 +751,71 @@ func (p *Parser) parseTTLClause(pos Pos, allowMultiValues bool) ([]*TTLExpr, err
 	return items, nil
 }
 
-func (p *Parser) parseTTLExpr(pos Pos) (*TTLExpr, error) {
-	columnExpr, err := p.parseExpr(pos)
-	if err != nil {
-		return nil, err
-	}
+func (p *Parser) tryParseTTLPolicy(pos Pos) (*TTLPolicy, error) {
+	var rule *TTLPolicyRule
 	switch {
-	case p.matchKeyword(KeywordDelete):
-		_ = p.lexer.consumeToken()
-	case p.matchKeyword(KeywordTo):
-		_ = p.lexer.consumeToken()
-		if p.tryConsumeKeyword(KeywordDisk) != nil || p.tryConsumeKeyword(KeywordVolume) != nil {
-			_, err := p.parseString(p.Pos())
+	case p.tryConsumeKeyword(KeywordTo) != nil:
+		if p.tryConsumeKeyword(KeywordDisk) != nil {
+			value, err := p.parseString(p.Pos())
 			if err != nil {
 				return nil, err
 			}
+			rule = &TTLPolicyRule{RulePos: pos, ToDisk: value}
+		} else if p.tryConsumeKeyword(KeywordVolume) != nil {
+			value, err := p.parseString(p.Pos())
+			if err != nil {
+				return nil, err
+			}
+			rule = &TTLPolicyRule{RulePos: pos, ToVolume: value}
 		} else {
-			return nil, fmt.Errorf("expected keyword <DISK> or <VOLUME>, but got %q", p.last().String)
+			return nil, fmt.Errorf("unexpected token: %q, expected DISK or VOLUME", p.lastTokenKind())
 		}
+	case p.matchKeyword(KeywordDelete), p.matchKeyword(KeywordRecompress):
+		token := p.last()
+		_ = p.lexer.consumeToken()
+		action := &TTLPolicyRuleAction{
+			ActionPos: token.Pos,
+			ActionEnd: token.End,
+			Action:    token.ToString(),
+		}
+		codec, err := p.tryParseCompressionCodecs(p.Pos())
+		if err != nil {
+			return nil, err
+		}
+		action.Codec = codec
+		rule = &TTLPolicyRule{RulePos: pos, Action: action}
+	default:
+		return nil, nil // nolint
+	}
+	policy := &TTLPolicy{Item: rule}
+
+	where, err := p.tryParseWhereClause(p.Pos())
+	if err != nil {
+		return nil, err
+	}
+	policy.Where = where
+
+	groupBy, err := p.tryParseGroupByClause(p.Pos())
+	if err != nil {
+		return nil, err
+	}
+	policy.GroupBy = groupBy
+	return policy, nil
+}
+
+func (p *Parser) parseTTLExpr(pos Pos) (*TTLExpr, error) {
+	columnExpr, err := p.parseExpr(pos)
+	if err != nil {
+		return nil, err
+	}
+	policy, err := p.tryParseTTLPolicy(p.Pos())
+	if err != nil {
+		return nil, err
 	}
 	return &TTLExpr{
 		TTLPos: pos,
 		Expr:   columnExpr,
+		Policy: policy,
 	}, nil
 }
 

parser/testdata/ddl/create_table_with_ttl_policy.sql

@@ -0,0 +1,33 @@
+CREATE TABLE tab
+(
+    d DateTime,
+    a Int
+)
+    ENGINE = MergeTree
+PARTITION BY toYYYYMM(d)
+ORDER BY d
+TTL d + INTERVAL 1 MONTH DELETE,
+    d + INTERVAL 1 WEEK TO VOLUME 'aaa',
+    d + INTERVAL 2 WEEK TO DISK 'bbb';
+
+
+CREATE TABLE table_with_where
+(
+    d DateTime,
+    a Int
+)
+    ENGINE = MergeTree
+PARTITION BY toYYYYMM(d)
+ORDER BY d
+TTL d + INTERVAL 1 MONTH DELETE WHERE toDayOfWeek(d) = 1;
+
+CREATE TABLE table_for_recompression
+(
+    d DateTime,
+    key UInt64,
+    value String
+) ENGINE MergeTree()
+ORDER BY tuple()
+PARTITION BY key
+TTL d + INTERVAL 1 MONTH RECOMPRESS CODEC(ZSTD(17)), d + INTERVAL 1 YEAR RECOMPRESS CODEC(LZ4HC(10))
+SETTINGS min_rows_for_wide_part = 0, min_bytes_for_wide_part = 0;

parser/testdata/ddl/format/create_mv_with_order_by.sql

@@ -13,5 +13,5 @@ AS
 SELECT * FROM test_table;
 
 -- Format SQL:
-CREATE MATERIALIZED VIEW IF NOT EXISTS test_mv ENGINE = ReplacingMergeTree() PRIMARY KEY (id) ORDER BY (id)  AS SELECT * FROM test_table;
+CREATE MATERIALIZED VIEW IF NOT EXISTS test_mv ENGINE = ReplacingMergeTree() PRIMARY KEY (id) ORDER BY (id) AS SELECT * FROM test_table;
 CREATE MATERIALIZED VIEW IF NOT EXISTS test_mv ENGINE = ReplacingMergeTree() PRIMARY KEY (id) AS SELECT * FROM test_table;