Skip to content

engines: ReplicatedReplacingMergeTree, ReplicatedMergeTree #2949

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
serprex merged 3 commits into from
May 28, 2025
Merged

engines: ReplicatedReplacingMergeTree, ReplicatedMergeTree #2949

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7f7a660
engines: ReplicatedReplacingMergeTree, ReplicatedMergeTree
serprex May 19, 2025
cef57e2
don't use {table} to avoid issues renaming (such as in resync)
serprex May 27, 2025
abdae82
Merge branch 'main' into rmt
serprex May 28, 2025
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
27 changes: 21 additions & 6 deletions flow/connectors/clickhouse/normalize.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,14 +159,29 @@ func generateCreateTableSQLForNormalizedTable(
}

var engine string
if tableMapping == nil {
tmEngine := protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE
if tableMapping != nil {
tmEngine = tableMapping.Engine
}
switch tmEngine {
case protos.TableEngine_CH_ENGINE_REPLACING_MERGE_TREE:
engine = fmt.Sprintf("ReplacingMergeTree(%s)", peerdb_clickhouse.QuoteIdentifier(versionColName))
} else if tableMapping.Engine == protos.TableEngine_CH_ENGINE_MERGE_TREE {
case protos.TableEngine_CH_ENGINE_MERGE_TREE:
engine = "MergeTree()"
} else if tableMapping.Engine == protos.TableEngine_CH_ENGINE_NULL {
case protos.TableEngine_CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE:
engine = fmt.Sprintf(
"ReplicatedReplacingMergeTree('/clickhouse/tables/{shard}/{database}/%s','{replica}',%s)",
peerdb_clickhouse.EscapeStr(tableIdentifier),
peerdb_clickhouse.QuoteIdentifier(versionColName),
)
case protos.TableEngine_CH_ENGINE_REPLICATED_MERGE_TREE:
engine = fmt.Sprintf(
"ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/%s','{replica}',%s)",
peerdb_clickhouse.EscapeStr(tableIdentifier),
peerdb_clickhouse.QuoteIdentifier(versionColName),
)
case protos.TableEngine_CH_ENGINE_NULL:
engine = "Null"
} else {
engine = fmt.Sprintf("ReplacingMergeTree(%s)", peerdb_clickhouse.QuoteIdentifier(versionColName))
}

// add sign and version columns
Expand All @@ -179,7 +194,7 @@ func generateCreateTableSQLForNormalizedTable(
orderByColumns = append([]string{sourceSchemaColName}, orderByColumns...)
}

if tableMapping == nil || tableMapping.Engine != protos.TableEngine_CH_ENGINE_NULL {
if tmEngine != protos.TableEngine_CH_ENGINE_NULL {
if len(orderByColumns) > 0 {
orderByStr := strings.Join(orderByColumns, ",")

Expand Down
2 changes: 1 addition & 1 deletion flow/shared/clickhouse/validation.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ import (
"go.temporal.io/sdk/log"
)

var acceptableTableEngines = []string{"ReplacingMergeTree", "MergeTree", "Null"}
var acceptableTableEngines = []string{"ReplacingMergeTree", "MergeTree", "ReplicatedReplacingMergeTree", "ReplicatedMergeTree", "Null"}

func CheckIfClickHouseCloudHasSharedMergeTreeEnabled(ctx context.Context, logger log.Logger,
conn clickhouse.Conn,
Expand Down
2 changes: 2 additions & 0 deletions protos/flow.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -251,6 +251,8 @@ enum TableEngine {
CH_ENGINE_REPLACING_MERGE_TREE = 0;
CH_ENGINE_MERGE_TREE = 1;
CH_ENGINE_NULL = 2;
CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE = 3;
CH_ENGINE_REPLICATED_MERGE_TREE = 4;
}

// protos for qrep
Expand Down
5 changes: 5 additions & 0 deletions ui/app/mirrors/create/cdc/schemabox.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,11 @@ export default function SchemaBox({
const engineOptions = [
{ value: 'CH_ENGINE_REPLACING_MERGE_TREE', label: 'ReplacingMergeTree' },
{ value: 'CH_ENGINE_MERGE_TREE', label: 'MergeTree' },
{
value: 'CH_ENGINE_REPLICATED_REPLACING_MERGE_TREE',
label: 'ReplicatedReplacingMergeTree',
},
{ value: 'CH_ENGINE_REPLICATED_MERGE_TREE', label: 'ReplicatedMergeTree' },
Comment on lines 249 to 252

@iamKunalGupta iamKunalGupta May 20, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

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

I'm thinking whether we should add a tooltip for new users.
Maybe simply saying that works with ch cloud and what works with oss clusters

@serprex serprex May 20, 2025

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.

mirror validation could check for host being CHC

@heavycrystal heavycrystal May 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

should be possible with CheckIfClickHouseCloudHasSharedMergeTreeEnabled

@serprex serprex May 20, 2025

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.

That only runs when we're in clickpipes. Validation here is for OSS peerdb users with CHC destination

@heavycrystal heavycrystal May 20, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

yes I mean we can use that function outside of current codepath to get if a service is SMT which should mean we disable these engines

@serprex serprex May 27, 2025

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.

I think for now we can omit this, people using CHC should know this engine doesn't apply to them

{ value: 'CH_ENGINE_NULL', label: 'Null' },
];

Expand Down