Skip to content
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

Clickhouse #397

Open
wants to merge 12 commits into
base: main
Choose a base branch
from
Open

Clickhouse #397

Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
c54bac4
do not instantiate munnecessary custom check runners
albertobruin Jan 9, 2025
7fedc08
add clickhouse package
albertobruin Jan 9, 2025
7f11dd6
add connection
albertobruin Jan 9, 2025
8f20db7
add new asset type
albertobruin Jan 9, 2025
506b25c
add config and basic db
albertobruin Jan 9, 2025
87bed6b
add dummy materializer
albertobruin Jan 9, 2025
cc0afe0
add some more stuff
albertobruin Jan 9, 2025
28c6f31
ensure checks sql works with clickhouse
albertobruin Jan 9, 2025
dd3c73d
move things around
albertobruin Jan 9, 2025
c5390d0
add clickhouse
albertobruin Jan 12, 2025
52df9a8
add missing mappings
albertobruin Jan 13, 2025
1abedc0
addapt to one query per statment, adjust materializations and tests
albertobruin Jan 13, 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
4 changes: 4 additions & 0 deletions .golangci.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,11 @@ linters-settings:
- databricks.Client
- athena.Client
- duckdb.DuckDBClient
- clickhouse.ClickHouseClient
- tea.Model
- driver.Rows
- driver.Row
- driver.Batch
goconst:
min-len: 5
min-occurrences: 5
Expand Down
2 changes: 2 additions & 0 deletions cmd/render.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,7 @@ import (
"github.com/alecthomas/chroma/v2/quick"
"github.com/bruin-data/bruin/pkg/athena"
"github.com/bruin-data/bruin/pkg/bigquery"
"github.com/bruin-data/bruin/pkg/clickhouse"
"github.com/bruin-data/bruin/pkg/config"
"github.com/bruin-data/bruin/pkg/databricks"
"github.com/bruin-data/bruin/pkg/date"
Expand Down Expand Up @@ -169,6 +170,7 @@ func Render() *cli.Command {
pipeline.AssetTypeSynapseQuery: synapse.NewRenderer(fullRefresh),
pipeline.AssetTypeAthenaQuery: athena.NewRenderer(fullRefresh, resultsLocation),
pipeline.AssetTypeDuckDBQuery: duck.NewMaterializer(fullRefresh),
pipeline.AssetTypeClickHouse: clickhouse.NewMaterializer(fullRefresh),
},
builder: DefaultPipelineBuilder,
writer: os.Stdout,
Expand Down
34 changes: 26 additions & 8 deletions cmd/run.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import (
"github.com/bruin-data/bruin/pkg/ansisql"
"github.com/bruin-data/bruin/pkg/athena"
"github.com/bruin-data/bruin/pkg/bigquery"
"github.com/bruin-data/bruin/pkg/clickhouse"
"github.com/bruin-data/bruin/pkg/config"
"github.com/bruin-data/bruin/pkg/connection"
"github.com/bruin-data/bruin/pkg/databricks"
Expand Down Expand Up @@ -769,39 +770,56 @@ func setupExecutors(

if s.WillRunTaskOfType(pipeline.AssetTypeAthenaQuery) || estimateCustomCheckType == pipeline.AssetTypeAthenaQuery || s.WillRunTaskOfType(pipeline.AssetTypeAthenaSeed) {
athenaOperator := athena.NewBasicOperator(conn, wholeFileExtractor, athena.NewMaterializer(fullRefresh))
athenaCustomCheckRunner := ansisql.NewCustomCheckOperator(conn, renderer)
athenaCheckRunner := athena.NewColumnCheckOperator(conn)

mainExecutors[pipeline.AssetTypeAthenaQuery][scheduler.TaskInstanceTypeMain] = athenaOperator
mainExecutors[pipeline.AssetTypeAthenaQuery][scheduler.TaskInstanceTypeColumnCheck] = athenaCheckRunner
mainExecutors[pipeline.AssetTypeAthenaQuery][scheduler.TaskInstanceTypeCustomCheck] = athenaCustomCheckRunner
mainExecutors[pipeline.AssetTypeAthenaQuery][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner

mainExecutors[pipeline.AssetTypeAthenaSeed][scheduler.TaskInstanceTypeMain] = seedOperator
mainExecutors[pipeline.AssetTypeAthenaSeed][scheduler.TaskInstanceTypeColumnCheck] = athenaCheckRunner
mainExecutors[pipeline.AssetTypeAthenaSeed][scheduler.TaskInstanceTypeCustomCheck] = athenaCustomCheckRunner
mainExecutors[pipeline.AssetTypeAthenaSeed][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner

if estimateCustomCheckType == pipeline.AssetTypeAthenaQuery {
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeColumnCheck] = athenaCheckRunner
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeCustomCheck] = athenaCustomCheckRunner
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner
}
}

if s.WillRunTaskOfType(pipeline.AssetTypeDuckDBQuery) || estimateCustomCheckType == pipeline.AssetTypeDuckDBQuery || s.WillRunTaskOfType(pipeline.AssetTypeDuckDBSeed) {
duckDBOperator := duck.NewBasicOperator(conn, wholeFileExtractor, duck.NewMaterializer(fullRefresh))
duckDBCustomCheckRunner := ansisql.NewCustomCheckOperator(conn, renderer)
duckDBCheckRunner := duck.NewColumnCheckOperator(conn)

mainExecutors[pipeline.AssetTypeDuckDBQuery][scheduler.TaskInstanceTypeMain] = duckDBOperator
mainExecutors[pipeline.AssetTypeDuckDBQuery][scheduler.TaskInstanceTypeColumnCheck] = duckDBCheckRunner
mainExecutors[pipeline.AssetTypeDuckDBQuery][scheduler.TaskInstanceTypeCustomCheck] = duckDBCustomCheckRunner
mainExecutors[pipeline.AssetTypeDuckDBQuery][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner

mainExecutors[pipeline.AssetTypeDuckDBSeed][scheduler.TaskInstanceTypeMain] = seedOperator
mainExecutors[pipeline.AssetTypeDuckDBSeed][scheduler.TaskInstanceTypeColumnCheck] = duckDBCheckRunner
mainExecutors[pipeline.AssetTypeDuckDBSeed][scheduler.TaskInstanceTypeCustomCheck] = duckDBCustomCheckRunner
mainExecutors[pipeline.AssetTypeDuckDBSeed][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner

if estimateCustomCheckType == pipeline.AssetTypeDuckDBQuery {
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeColumnCheck] = duckDBCheckRunner
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeCustomCheck] = duckDBCustomCheckRunner
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner
}
}

// ClickHouse
if s.WillRunTaskOfType(pipeline.AssetTypeClickHouse) || estimateCustomCheckType == pipeline.AssetTypeClickHouse || s.WillRunTaskOfType(pipeline.AssetTypeClickHouseSeed) {
clickHouseOperator := clickhouse.NewBasicOperator(conn, wholeFileExtractor, clickhouse.NewMaterializer(fullRefresh))
checkRunner := clickhouse.NewColumnCheckOperator(conn)

mainExecutors[pipeline.AssetTypeClickHouse][scheduler.TaskInstanceTypeMain] = clickHouseOperator
mainExecutors[pipeline.AssetTypeClickHouse][scheduler.TaskInstanceTypeColumnCheck] = checkRunner
mainExecutors[pipeline.AssetTypeClickHouse][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner

mainExecutors[pipeline.AssetTypeClickHouseSeed][scheduler.TaskInstanceTypeMain] = seedOperator
mainExecutors[pipeline.AssetTypeClickHouseSeed][scheduler.TaskInstanceTypeColumnCheck] = checkRunner
mainExecutors[pipeline.AssetTypeClickHouseSeed][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner

if estimateCustomCheckType == pipeline.AssetTypeClickHouse {
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeColumnCheck] = checkRunner
mainExecutors[pipeline.AssetTypePython][scheduler.TaskInstanceTypeCustomCheck] = customCheckRunner
}
}

Expand Down
34 changes: 21 additions & 13 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,12 @@
module github.com/bruin-data/bruin

go 1.22
go 1.22.0

toolchain go1.22.2
toolchain go1.23.2

require (
cloud.google.com/go/bigquery v1.60.0
github.com/ClickHouse/clickhouse-go/v2 v2.30.0
github.com/DATA-DOG/go-sqlmock v1.5.0
github.com/alecthomas/chroma/v2 v2.13.0
github.com/charmbracelet/bubbletea v1.1.2
Expand Down Expand Up @@ -33,7 +34,7 @@ require (
github.com/snowflakedb/gosnowflake v1.8.0
github.com/sourcegraph/conc v0.3.0
github.com/spf13/afero v1.11.0
github.com/stretchr/testify v1.9.0
github.com/stretchr/testify v1.10.0
github.com/uber/athenadriver v1.1.15
github.com/urfave/cli/v2 v2.27.1
github.com/xlab/treeprint v1.2.0
Expand All @@ -57,8 +58,9 @@ require (
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.10.0 // indirect
github.com/Azure/azure-sdk-for-go/sdk/internal v1.5.2 // indirect
github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.3.1 // indirect
github.com/ClickHouse/ch-go v0.63.1 // indirect
github.com/JohnCGriffin/overflow v0.0.0-20211019200055-46fa312c352c // indirect
github.com/andybalholm/brotli v1.1.0 // indirect
github.com/andybalholm/brotli v1.1.1 // indirect
github.com/apache/arrow/go/v12 v12.0.1 // indirect
github.com/apache/arrow/go/v14 v14.0.2 // indirect
github.com/apache/arrow/go/v17 v17.0.0 // indirect
Expand Down Expand Up @@ -97,8 +99,10 @@ require (
github.com/form3tech-oss/jwt-go v3.2.5+incompatible // indirect
github.com/fsnotify/fsnotify v1.5.4 // indirect
github.com/gabriel-vasile/mimetype v1.4.3 // indirect
github.com/go-faster/city v1.0.1 // indirect
github.com/go-faster/errors v0.7.1 // indirect
github.com/go-jose/go-jose/v3 v3.0.3 // indirect
github.com/go-logr/logr v1.4.1 // indirect
github.com/go-logr/logr v1.4.2 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/go-openapi/jsonpointer v0.19.5 // indirect
github.com/go-openapi/swag v0.21.1 // indirect
Expand All @@ -120,12 +124,12 @@ require (
github.com/hashicorp/go-retryablehttp v0.7.7 // indirect
github.com/jackc/pgpassfile v1.0.0 // indirect
github.com/jackc/pgservicefile v0.0.0-20231201235250-de7065d80cb9 // indirect
github.com/jackc/puddle/v2 v2.2.1 // indirect
github.com/jackc/puddle/v2 v2.2.2 // indirect
github.com/jmespath/go-jmespath v0.4.0 // indirect
github.com/josharian/intern v1.0.0 // indirect
github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/asmfmt v1.3.2 // indirect
github.com/klauspost/compress v1.17.9 // indirect
github.com/klauspost/compress v1.17.11 // indirect
github.com/klauspost/cpuid/v2 v2.2.8 // indirect
github.com/lucasb-eyer/go-colorful v1.2.0 // indirect
github.com/mailru/easyjson v0.7.7 // indirect
Expand All @@ -142,14 +146,17 @@ require (
github.com/muesli/ansi v0.0.0-20230316100256-276c6243b2f6 // indirect
github.com/muesli/cancelreader v0.2.2 // indirect
github.com/muesli/termenv v0.15.2 // indirect
github.com/pierrec/lz4/v4 v4.1.21 // indirect
github.com/paulmach/orb v0.11.1 // indirect
github.com/pierrec/lz4/v4 v4.1.22 // indirect
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c // indirect
github.com/pmezard/go-difflib v1.0.0 // indirect
github.com/rivo/uniseg v0.4.7 // indirect
github.com/rogpeppe/go-internal v1.12.0 // indirect
github.com/rogpeppe/go-internal v1.13.1 // indirect
github.com/rs/zerolog v1.28.0 // indirect
github.com/russross/blackfriday/v2 v2.1.0 // indirect
github.com/segmentio/asm v1.2.0 // indirect
github.com/segmentio/backo-go v1.1.0 // indirect
github.com/shopspring/decimal v1.4.0 // indirect
github.com/sirupsen/logrus v1.9.3 // indirect
github.com/stretchr/objx v0.5.2 // indirect
github.com/tidwall/gjson v1.14.4 // indirect
Expand All @@ -161,16 +168,17 @@ require (
github.com/xwb1989/sqlparser v0.0.0-20180606152119-120387863bf2 // indirect
github.com/zeebo/xxh3 v1.0.2 // indirect
go.opencensus.io v0.24.0 // indirect
go.opentelemetry.io/auto/sdk v1.1.0 // indirect
go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.49.0 // indirect
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.49.0 // indirect
go.opentelemetry.io/otel v1.24.0 // indirect
go.opentelemetry.io/otel/metric v1.24.0 // indirect
go.opentelemetry.io/otel/trace v1.24.0 // indirect
go.opentelemetry.io/otel v1.33.0 // indirect
go.opentelemetry.io/otel/metric v1.33.0 // indirect
go.opentelemetry.io/otel/trace v1.33.0 // indirect
go.uber.org/multierr v1.11.0 // indirect
golang.org/x/crypto v0.31.0 // indirect
golang.org/x/mod v0.19.0 // indirect
golang.org/x/net v0.33.0 // indirect
golang.org/x/sys v0.28.0 // indirect
golang.org/x/sys v0.29.0 // indirect
golang.org/x/term v0.27.0 // indirect
golang.org/x/text v0.21.0 // indirect
golang.org/x/time v0.5.0 // indirect
Expand Down
Loading
Loading