diff --git a/cmd/sling/tests/replications/r.25.prometheus_buffer.yaml b/cmd/sling/tests/replications/r.25.prometheus_buffer.yaml index 646662d29..6bb5fa7a4 100644 --- a/cmd/sling/tests/replications/r.25.prometheus_buffer.yaml +++ b/cmd/sling/tests/replications/r.25.prometheus_buffer.yaml @@ -4,11 +4,14 @@ target: local defaults: mode: full-refresh target_options: - format: csv + format: parquet file_max_rows: 10000 streams: prometheus_metrics: + transforms: + - '*': cast(value, "string") + - event_date: date_format(now(), "%Y-%m-%d") sql: | - up#{"start": "now-30m", "end": "now", "step": "5m"} - object: '/tmp/prometheus_test_{run_timestamp}.csv' + sum(go_gc_duration_seconds) by (job, instance, quantile) # {"start": "now-2M"} + object: '/tmp/output/prometheus_test.parquet' diff --git a/cmd/sling/tests/replications/r.44.staged_transforms_test.yaml b/cmd/sling/tests/replications/r.44.staged_transforms_test.yaml index 6e9bcfbf5..98e0c9c2c 100644 --- a/cmd/sling/tests/replications/r.44.staged_transforms_test.yaml +++ b/cmd/sling/tests/replications/r.44.staged_transforms_test.yaml @@ -87,6 +87,7 @@ streams: - "*": cast(value, "string") - "*": remove_diacritics(value) - calculated_amount: 'cast(value, "decimal")' + event_date: date_format(now(), "%Y-%m-%d") hooks: post: @@ -102,6 +103,7 @@ streams: full_name, email_hash, calculated_amount, + event_date, category, display_name, processed_status @@ -126,6 +128,12 @@ streams: - type: check check: store.result[0].email == "john@example.com" # lowercased + - type: check + check: length(store.result[0].email_hash) > 30 + + - type: check + check: length(store.result[0].event_date) == 10 + - type: check check: store.result[0].full_name == "John Doe" # record reference combining trimmed names diff --git a/cmd/sling/tests/replications/r.54.mongo_objectid_filter.yaml b/cmd/sling/tests/replications/r.54.mongo_objectid_filter.yaml new file mode 100644 index 000000000..fc01448a8 --- /dev/null +++ b/cmd/sling/tests/replications/r.54.mongo_objectid_filter.yaml @@ -0,0 +1,39 @@ +source: mongo +target: postgres + +defaults: + mode: full-refresh + source_options: + flatten: 1 + +hooks: + start: + # Clean up test table if exists + - type: query + connection: '{target.name}' + query: DROP TABLE IF EXISTS public.mongo_objectid_test; + + end: + # Verify records were inserted + - type: query + connection: '{target.name}' + query: SELECT COUNT(*) as count FROM public.mongo_objectid_test + into: result + + - type: check + check: int_parse(store.result[0].count) == 90 + failure_message: Should have 90 rows + + # Clean up test table + - type: query + connection: '{target.name}' + query: DROP TABLE IF EXISTS public.mongo_objectid_test; + +streams: + # Test ObjectID filtering on existing collection + # This tests that ObjectID hex strings in filters are properly converted + default.test1k_mongodb: + object: public.mongo_objectid_test + # Filter using ObjectID range - these will be converted to proper ObjectID types + # Using a range that should capture some documents + where: '{"_id": {"$gte": "67859d8ee682ab32317abc6f", "$lte": "67859d8ee682ab32317abcc8"}}' \ No newline at end of file diff --git a/cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml b/cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml new file mode 100644 index 000000000..1e892c1d5 --- /dev/null +++ b/cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml @@ -0,0 +1,41 @@ +source: LOCAL +target: SNOWFLAKE + +hooks: + start: + - type: command + command: bash cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml.prep.sh + + - type: query + connection: '{target.name}' + query: DROP TABLE IF EXISTS public.sling_nested_json; + + end: + # Verify total records were imported + - type: query + connection: '{target.name}' + query: SELECT COUNT(*) as count FROM public.sling_nested_json + into: result + + - type: check + check: int_parse(store.result[0].count) == 1700 + success_message: "✓ All 1700 JSON records imported successfully (1200 from nested1.json + 500 from nested2.json)" + + # Cleanup files + - type: command + command: rm -rf /tmp/sling-test-files/ + + # Cleanup table + - type: query + connection: '{target.name}' + query: DROP TABLE IF EXISTS public.sling_nested_json; + +streams: + file:///tmp/sling-test-files/*.json: + object: public.sling_nested_json + mode: full-refresh + source_options: + flatten: 1 + +env: + SLING_SAMPLE_SIZE: 100 \ No newline at end of file diff --git a/cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml.prep.sh b/cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml.prep.sh new file mode 100644 index 000000000..0e7fe2b32 --- /dev/null +++ b/cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml.prep.sh @@ -0,0 +1,26 @@ +mkdir -p /tmp/sling-test-files + +# Create nested1.json with 1200 records (2 columns each) +echo '[' > /tmp/sling-test-files/nested1.json +for i in {1..1200}; do + if [ $i -gt 1 ]; then + echo "," >> /tmp/sling-test-files/nested1.json + fi + if [ $i -gt 1100 ]; then + echo -n "{\"id\": $i, \"name\": \"User$i\", \"email\": \"user$(($i))@example.com\"}" >> /tmp/sling-test-files/nested1.json + else + echo -n "{\"id\": $i, \"name\": \"User$i\"}" >> /tmp/sling-test-files/nested1.json + fi +done +echo -e "\n]" >> /tmp/sling-test-files/nested1.json + +# Create nested2.json with 500 records (3 columns each) +echo '[' > /tmp/sling-test-files/nested2.json +for i in {1..500}; do + if [ $i -gt 1 ]; then + echo "," >> /tmp/sling-test-files/nested2.json + fi + value=$((i * 100)) + echo -n "{\"id\": $((1200 + i)), \"email\": \"user$((1200 + i))@example.com\", \"value\": $value}" >> /tmp/sling-test-files/nested2.json +done +echo -e "\n]" >> /tmp/sling-test-files/nested2.json \ No newline at end of file diff --git a/cmd/sling/tests/replications/r.56.mssql_starrocks_money.yaml b/cmd/sling/tests/replications/r.56.mssql_starrocks_money.yaml new file mode 100644 index 000000000..d0ab3362b --- /dev/null +++ b/cmd/sling/tests/replications/r.56.mssql_starrocks_money.yaml @@ -0,0 +1,220 @@ +source: mssql +target: starrocks + +defaults: + mode: full-refresh + +hooks: + start: + - type: query + connection: '{source.name}' + query: | + -- Drop temp table if it exists + IF OBJECT_ID('dbo.money_test', 'U') IS NOT NULL DROP TABLE dbo.money_test; + + -- Create table with money and decimal columns + CREATE TABLE dbo.money_test ( + id INT PRIMARY KEY, + description VARCHAR(100), + money_col MONEY, + smallmoney_col SMALLMONEY, + decimal_col DECIMAL(24,6), + decimal_high_precision DECIMAL(38,10) + ); + + -- Insert test data + INSERT INTO dbo.money_test (id, description, money_col, smallmoney_col, decimal_col, decimal_high_precision) VALUES + (1, 'Small positive', 4.00, 4.00, 4.0000, 4.0000000000), + (2, 'Problematic value', 4491.00, 4491.00, 4491.0000, 4491.0000000000), + (3, 'Large value', 99999.99, 99999.99, 99999.9900, 99999.9900000000), + (4, 'Negative value', -1234.56, -1234.56, -1234.5600, -1234.5600000000), + (5, 'Zero', 0.00, 0.00, 0.0000, 0.0000000000), + (6, 'Small decimal', 0.01, 0.01, 0.0100, 0.0100000000), + (7, 'NULL values', NULL, NULL, NULL, NULL), + (8, 'Max smallmoney', 214748.3647, 214748.3647, 214748.3647, 214748.3647000000), + (9, 'Large money', 922337203685.4775, NULL, 922337203685.4775, 922337203685.4775000000), + (10, 'Precision test', 123.4567, 123.4567, 123.4567, 123.4567890123); + + end: + - type: check + check: runs.dbo_money_test.status == "success" + on_failure: break + + # Get source aggregations + - type: query + connection: '{source.name}' + query: | + SELECT + COUNT(*) as row_count, + SUM(CAST(money_col AS DECIMAL(38,4))) as money_sum, + SUM(CAST(smallmoney_col AS DECIMAL(38,4))) as smallmoney_sum, + SUM(decimal_col) as decimal_sum, + MIN(CAST(money_col AS DECIMAL(38,4))) as money_min, + MAX(CAST(money_col AS DECIMAL(38,4))) as money_max + FROM dbo.money_test + into: source_aggregations + + # Get target aggregations + - type: query + connection: '{target.name}' + query: | + SELECT + COUNT(*) as row_count, + SUM(money_col) as money_sum, + SUM(smallmoney_col) as smallmoney_sum, + SUM(decimal_col) as decimal_sum, + MIN(money_col) as money_min, + MAX(money_col) as money_max + FROM public.money_test + into: target_aggregations + + - type: log + message: | + + Source (MSSQL) aggregations: + {pretty_table(store.source_aggregations)} + + Target (StarRocks) aggregations: + {pretty_table(store.target_aggregations)} + + # Check row count + - type: check + check: int_parse(store.source_aggregations[0].row_count) == int_parse(store.target_aggregations[0].row_count) + failure_message: Row count mismatch. Source {store.source_aggregations[0].row_count} != Target {store.target_aggregations[0].row_count} + + # Check money_col sum + - type: check + check: float_parse(store.source_aggregations[0].money_sum) == float_parse(store.target_aggregations[0].money_sum) + failure_message: Money column sum mismatch. Source {store.source_aggregations[0].money_sum} != Target {store.target_aggregations[0].money_sum} + + # Check decimal_col sum + - type: check + check: float_parse(store.source_aggregations[0].decimal_sum) == float_parse(store.target_aggregations[0].decimal_sum) + failure_message: Decimal column sum mismatch. Source {store.source_aggregations[0].decimal_sum} != Target {store.target_aggregations[0].decimal_sum} + + # Get individual rows to check specific values + - type: query + connection: '{target.name}' + query: | + SELECT + id, + description, + money_col, + smallmoney_col, + decimal_col, + decimal_high_precision + FROM public.money_test + WHERE id IN (2, 9) + ORDER BY id + into: target_specific_rows + + - type: log + message: | + Target specific rows (checking problematic value 4491.00): + {pretty_table(store.target_specific_rows)} + + # Check the problematic value 4491.00 + - type: check + check: float_parse(store.target_specific_rows[0].money_col) == 4491.00 + failure_message: Problematic value 4491.00 not preserved correctly. Got {store.target_specific_rows[0].money_col} + + # Check large money value + - type: check + check: float_parse(store.target_specific_rows[1].money_col) == 922337203685.4775 + failure_message: Large money value not preserved correctly. Got {store.target_specific_rows[1].money_col} + + # Get source rows for comparison + - type: query + connection: '{source.name}' + query: | + SELECT + id, + description, + CAST(money_col AS DECIMAL(38,4)) as money_value, + CAST(decimal_col AS DECIMAL(38,4)) as decimal_value + FROM dbo.money_test + ORDER BY id + into: source_rows + + # Get target rows for comparison + - type: query + connection: '{target.name}' + query: | + SELECT + id, + description, + money_col as money_value, + decimal_col as decimal_value + FROM public.money_test + ORDER BY id + into: target_rows + - type: query + connection: '{target.name}' + query: | + SELECT + id, + description, + money_col as money_value, + decimal_col as decimal_value + FROM public.money_test_sql + ORDER BY id + into: target_rows_sql + + - type: log + message: | + Row-by-row comparison: + + Source (MSSQL) rows: + {pretty_table(store.source_rows)} + + Target (StarRocks) rows: + {pretty_table(store.target_rows)} + + Target (SQL) (StarRocks) rows: + {pretty_table(store.target_rows_sql)} + + # Check specific rows match + - type: check + check: float_parse(store.source_rows[0].money_value) == float_parse(store.target_rows[0].money_value) + failure_message: Row 1 money value mismatch. Source {store.source_rows[0].money_value} != Target {store.target_rows[0].money_value} + + - type: check + check: float_parse(store.source_rows[1].money_value) == float_parse(store.target_rows[1].money_value) + failure_message: Row 2 (4491.00) money value mismatch. Source {store.source_rows[1].money_value} != Target {store.target_rows[1].money_value} + + - type: check + check: float_parse(store.source_rows[1].money_value) == float_parse(store.target_rows_sql[1].money_value) + failure_message: Row 2 (4491.00) money value mismatch. Source {store.source_rows[1].money_value} != Target {store.target_rows_sql[1].money_value} + + - type: check + check: float_parse(store.source_rows[8].money_value) == float_parse(store.target_rows[8].money_value) + failure_message: Row 9 (large value) money value mismatch. Source {store.source_rows[8].money_value} != Target {store.target_rows[8].money_value} + + - type: check + check: float_parse(store.source_rows[8].decimal_value) == float_parse(store.target_rows_sql[8].decimal_value) + failure_message: Row 9 (large value) money value mismatch. Source {store.source_rows[8].decimal_value} != Target {store.target_rows_sql[8].decimal_value} + + # Cleanup + - type: query + connection: '{source.name}' + query: DROP TABLE IF EXISTS dbo.money_test + + - type: query + connection: '{target.name}' + query: | + DROP TABLE IF EXISTS public.money_test; + DROP TABLE IF EXISTS public.money_test_sql; + +streams: + dbo.money_test: + object: public.money_test + mode: full-refresh + + dbo.money_test_sql: + sql: | + select * from dbo.money_test + object: public.money_test_sql + mode: full-refresh + +env: + SLING_SAMPLE_SIZE: 2 \ No newline at end of file diff --git a/cmd/sling/tests/suite.cli.yaml b/cmd/sling/tests/suite.cli.yaml index 3da79858c..9c3535b8c 100644 --- a/cmd/sling/tests/suite.cli.yaml +++ b/cmd/sling/tests/suite.cli.yaml @@ -199,7 +199,8 @@ - id: 34 name: Run sling with Prometheus source and custom query - run: 'sling run --src-conn prometheus --src-stream ''sum(go_gc_duration_seconds) by (job, instance, quantile) # {"start": "now-2M"}'' --stdout -d' + run: > + sling run --src-conn prometheus --src-stream 'sum(go_gc_duration_seconds) by (job, instance, quantile) # {"start": "now-2M"}' --stdout -d output_contains: - quantile @@ -576,6 +577,9 @@ min_rows: 1 output_contains: - 'using range' + - "changing column type via transform for 'timestamp': bigint => string" + - "changing column type via transform for 'value': decimal => string" + - "'timestamp':'text', 'value':'text'" # the timestamp & value columns were successfully casted to string - id: 81 name: Prometheus issue 551 (https://github.com/slingdata-io/sling-cli/issues/551) @@ -945,3 +949,23 @@ rows: 0 output_contains: - 'execution succeeded' + +# Test MongoDB ObjectID filtering +- id: 116 + name: MongoDB ObjectID filter test + run: 'sling run -r cmd/sling/tests/replications/r.54.mongo_objectid_filter.yaml' + +- id: 117 + name: JSON to Snowflake nested import test + run: 'sling run -d -r cmd/sling/tests/replications/r.55.json_snowflake_nested.yaml' + +# Test SQL Server money type to StarRocks +- id: 118 + name: SQL Server money type to StarRocks + run: 'sling run -d -r cmd/sling/tests/replications/r.56.mssql_starrocks_money.yaml' + conns: + - mssql + - starrocks + output_contains: + - 'execution succeeded' + - 'Row-by-row comparison' diff --git a/core/dbio/api/api.go b/core/dbio/api/api.go index 2755faed8..17ce30128 100644 --- a/core/dbio/api/api.go +++ b/core/dbio/api/api.go @@ -394,6 +394,9 @@ var ( runSequence = func(s Sequence, ep *Endpoint) (err error) { return g.Error("please use the official sling-cli release for running API sequences") } + FetchSpec = func(_ string) (string, error) { + return "", g.Error("please use the official sling-cli release for fetching API specs") + } ) // RegisterQueue creates a new queue with the given name diff --git a/core/dbio/connection/connection.go b/core/dbio/connection/connection.go index 95d725e53..cd65b638c 100644 --- a/core/dbio/connection/connection.go +++ b/core/dbio/connection/connection.go @@ -559,6 +559,9 @@ func (c *Connection) setURL() (err error) { if _, ok := c.Data["role"]; ok { template = template + "&role={role}" } + if _, ok := c.Data["statement_timeout"]; ok { + template = template + "&statement_timeout={statement_timeout}" + } case dbio.TypeDbRedshift: setIfMissing("username", c.Data["user"]) setIfMissing("password", "") @@ -1229,9 +1232,9 @@ func ParseLocation(location string) (conn Connection, objectExpr string, err err } // LoadAPISpec loads the spec from the spec location -func LoadAPISpec(specLocation string) (spec api.Spec, err error) { +func LoadAPISpec(specIdentifier string) (spec api.Spec, err error) { - if specLocation == "" { + if specIdentifier == "" { return spec, g.Error("invalid or missing spec") } @@ -1323,18 +1326,18 @@ func LoadAPISpec(specLocation string) (spec api.Spec, err error) { var specBody string switch { // load from location - case strings.HasPrefix(specLocation, "file://"): - specPath := strings.TrimPrefix(specLocation, "file://") + case strings.HasPrefix(specIdentifier, "file://"): + specPath := strings.TrimPrefix(specIdentifier, "file://") bytes, err := os.ReadFile(specPath) if err != nil { return spec, g.Error(err, "could not read api spec from: %s", specPath) } specBody = string(bytes) - case strings.HasPrefix(specLocation, "https://"), strings.HasPrefix(specLocation, "http://"): + case strings.HasPrefix(specIdentifier, "https://"), strings.HasPrefix(specIdentifier, "http://"): // download raw http - specURL, err := transformToRawGitURL(specLocation) + specURL, err := transformToRawGitURL(specIdentifier) if err != nil { - return spec, g.Error(err, "could not make git spec URL for download: %s", specLocation) + return spec, g.Error(err, "could not make git spec URL for download: %s", specIdentifier) } redirect: resp, respBytes, err := net.ClientDo("GET", specURL, nil, nil) @@ -1348,15 +1351,15 @@ func LoadAPISpec(specLocation string) (spec api.Spec, err error) { } } specBody = string(respBytes) - default: + case strings.Contains(specIdentifier, "/"): // connect to location and download - specConn, specPath, err := ParseLocation(specLocation) + specConn, specPath, err := ParseLocation(specIdentifier) if err != nil { - return spec, g.Error(err, "could not read api spec from location: %s", specLocation) + return spec, g.Error(err, "could not read api spec from location: %s", specIdentifier) } fc, err := specConn.AsFile() if err != nil { - return spec, g.Error(err, "could not connection to location: %s", specLocation) + return spec, g.Error(err, "could not connection to location: %s", specIdentifier) } reader, err := fc.GetReader(specPath) @@ -1369,12 +1372,17 @@ func LoadAPISpec(specLocation string) (spec api.Spec, err error) { return spec, g.Error(err, "could not read api spec from: %s", specPath) } specBody = string(bytes) + default: + specBody, err = api.FetchSpec(specIdentifier) + if err != nil { + return spec, g.Error(err, "could not fetch api spec: %s", specIdentifier) + } } // load spec spec, err = api.LoadSpec(specBody) if err != nil { - return spec, g.Error(err, "could not load spec from %s", specLocation) + return spec, g.Error(err, "could not load spec from %s", specIdentifier) } return diff --git a/core/dbio/database/database.go b/core/dbio/database/database.go index 042079f6f..175d67e6d 100755 --- a/core/dbio/database/database.go +++ b/core/dbio/database/database.go @@ -1483,7 +1483,9 @@ func SQLColumns(colTypes []ColumnType, conn Connection) (columns iop.Columns) { col.DbScale = colType.Scale col.Stats.MaxDecLen = colType.Scale } - if col.DbPrecision > 0 { + + // some instances where the precision is returned too small. + if col.DbPrecision > env.DdlMinDecLength { if g.In(conn.GetType(), dbio.TypeDbOracle) { // only mark as sourced is scale is specified // https://github.com/slingdata-io/sling-cli/issues/584 @@ -1491,6 +1493,8 @@ func SQLColumns(colTypes []ColumnType, conn Connection) (columns iop.Columns) { } else { col.Sourced = true } + } else { + col.Sourced = false } } @@ -2577,7 +2581,7 @@ func (conn *BaseConn) Merge(srcTable string, tgtTable string, primKeys []string) cnt, err = Merge(conn.Self(), nil, srcTable, tgtTable, primKeys) } if err != nil { - err = g.Error(err, "could not upsert") + err = g.Error(err, "could not merge") } return cast.ToInt64(cnt), err } @@ -2702,6 +2706,7 @@ func (conn *BaseConn) GenerateMergeConfig(srcTable string, tgtTable string, pkFi tgtFields := conn.Type.QuoteNames(tgtCols.Names()...) setFields := []string{} + setFieldsAll := []string{} insertFields := []string{} placeholderFields := []string{} for _, tgtColName := range tgtCols.Names() { @@ -2721,14 +2726,21 @@ func (conn *BaseConn) GenerateMergeConfig(srcTable string, tgtTable string, pkFi phExpr := strings.ReplaceAll(colExpr, srcColNameQ, g.F("ph.%s", srcColNameQ)) placeholderFields = append(placeholderFields, phExpr) + + setSrcExpr := strings.ReplaceAll(colExpr, srcColNameQ, g.F("src.%s", srcColNameQ)) + setField := g.F("%s = %s", tgtColNameQ, setSrcExpr) + setFieldsAll = append(setFieldsAll, setField) if _, ok := pkFieldMap[tgtCol.Name]; !ok { // is not a pk field - setSrcExpr := strings.ReplaceAll(colExpr, srcColNameQ, g.F("src.%s", srcColNameQ)) - setField := g.F("%s = %s", tgtColNameQ, setSrcExpr) setFields = append(setFields, setField) } } + // if PK is all the available columns + if len(setFields) == 0 && len(setFieldsAll) > 0 { + setFields = setFieldsAll + } + // cast into the correct type srcFields := conn.Self().CastColumnsForSelect(srcColumns, tgtColumns) diff --git a/core/dbio/database/database_clickhouse.go b/core/dbio/database/database_clickhouse.go index 9157cd685..a35981352 100755 --- a/core/dbio/database/database_clickhouse.go +++ b/core/dbio/database/database_clickhouse.go @@ -27,7 +27,8 @@ import ( // ClickhouseConn is a Clikchouse connection type ClickhouseConn struct { BaseConn - URL string + URL string + version string } // Init initiates the object @@ -41,6 +42,27 @@ func (conn *ClickhouseConn) Init() error { return conn.BaseConn.Init() } +func (conn *ClickhouseConn) getVersion() { + data, err := conn.Query(`SELECT version()` + env.NoDebugKey) + if err != nil { + conn.version = "21.0" + } else if len(data.Rows) > 0 { + conn.version = cast.ToString(data.Rows[0][0]) + } +} + +func (conn *ClickhouseConn) Version() int { + parts := strings.Split(conn.version, ".") + if len(parts) > 0 { + v := cast.ToInt(parts[0]) + if v == 0 { + v = 21 + } + return v + } + return 21 +} + func (conn *ClickhouseConn) Connect(timeOut ...int) (err error) { tlsConfig, err := conn.makeTlsConfig() if err != nil { @@ -54,6 +76,8 @@ func (conn *ClickhouseConn) Connect(timeOut ...int) (err error) { if strings.Contains(err.Error(), "unexpected packet") { g.Info(env.MagentaString("Try using the `http_url` instead to connect to Clickhouse via HTTP. See https://docs.slingdata.io/connections/database-connections/clickhouse")) } + } else { + conn.getVersion() } return err } @@ -246,6 +270,8 @@ func (conn *ClickhouseConn) Connect(timeOut ...int) (err error) { if strings.Contains(err.Error(), "unexpected packet") { g.Info(env.MagentaString("Try using the `http_url` instead to connect to Clickhouse via HTTP. See https://docs.slingdata.io/connections/database-connections/clickhouse")) } + } else { + conn.getVersion() } return err @@ -549,6 +575,33 @@ func (conn *ClickhouseConn) GenerateInsertStatement(tableName string, cols iop.C return statement } +// Merge inserts / updates from a srcTable into a target table. +// Assuming the srcTable has some or all of the tgtTable fields with matching types +func (conn *ClickhouseConn) Merge(srcTable string, tgtTable string, primKeys []string) (rowAffCnt int64, err error) { + maxTries := 10 + tryCount := 0 + +retry: + tryCount++ + rowAffCnt, err = conn.BaseConn.Merge(srcTable, tgtTable, primKeys) + if err != nil && strings.Contains(err.Error(), "Too many unfinished mutations") && tryCount <= maxTries { + g.Warn("got error: `Too many unfinished mutations`. Retrying after 5 seconds... (attempt %d/%d)", tryCount, maxTries) + time.Sleep(5 * time.Second) + + // re-open transaction + conn.Rollback() + txOptions := sql.TxOptions{Isolation: sql.LevelDefault} + if err = conn.BeginContext(conn.context.Ctx, &txOptions); err != nil { + err = g.Error(err, "could not re-open transaction to complete merge") + return 0, err + } + + goto retry + } + + return rowAffCnt, err +} + // GenerateMergeSQL generates the upsert SQL func (conn *ClickhouseConn) GenerateMergeSQL(srcTable string, tgtTable string, pkFields []string) (sql string, err error) { upsertMap, err := conn.BaseConn.GenerateMergeExpressions(srcTable, tgtTable, pkFields) @@ -570,6 +623,25 @@ func (conn *ClickhouseConn) GenerateMergeSQL(srcTable string, tgtTable string, p select {src_fields} from {src_table} src ` + + if conn.Version() >= 23 { + // use lightweight delete + // see https://github.com/slingdata-io/sling-cli/issues/593 + sqlTempl = ` + delete from {tgt_table} + where ({tgt_pk_fields}) in ( + select {src_pk_fields} + from {src_table} src + ) + ; + + insert into {tgt_table} + ({insert_fields}) + select {src_fields} + from {src_table} src + ` + } + sql = g.R( sqlTempl, "src_table", srcTable, diff --git a/core/dbio/database/database_databricks.go b/core/dbio/database/database_databricks.go index 59b6a4d58..0a81452b3 100644 --- a/core/dbio/database/database_databricks.go +++ b/core/dbio/database/database_databricks.go @@ -765,6 +765,7 @@ func (conn *DatabricksConn) GenerateMergeSQL(srcTable string, tgtTable string, p // getOrCreateVolume creates a volume if it doesn't exist, similar to Snowflake's getOrCreateStage func (conn *DatabricksConn) getOrCreateVolume(schema string) (internalVolume string, err error) { + var volume Table internalVolume = conn.GetProp("internal_volume") if internalVolume == "" { @@ -776,23 +777,30 @@ func (conn *DatabricksConn) getOrCreateVolume(schema string) (internalVolume str } // Create volume name similar to how Snowflake creates stages - volumeName := "sling_volume" - volumeFullName := fmt.Sprintf("%s.%s.%s", conn.Catalog, schema, volumeName) + volume = Table{ + Database: conn.Catalog, + Schema: schema, + Name: "sling_volume", + Dialect: dbio.TypeDbDatabricks, + } - sql := g.R( - conn.template.Core["create_volume"], - "volume_name", volumeFullName, - ) - _, err := conn.Exec(sql + env.NoDebugKey) - if err != nil { + volumeFullName := volume.FDQN() + sql := g.R(conn.template.Core["create_volume"], "volume_name", volumeFullName) + if _, err = conn.Exec(sql + env.NoDebugKey); err != nil { return "", g.Error(err, "could not create volume: %s", volumeFullName) } - - volumePath := fmt.Sprintf("/Volumes/%s/%s/%s", conn.Catalog, schema, volumeName) - conn.SetProp("internal_volume", volumePath) - internalVolume = volumeFullName + conn.SetProp("internal_volume", volumeFullName) + } else { + volume, err = ParseTableName(internalVolume, dbio.TypeDbDatabricks) + if err != nil { + return "", g.Error(err, "invalid volume name, should be in format: `catalog_name`.`schema_name`.`volume_name`") + } } - return internalVolume, nil + + volumePath := fmt.Sprintf("/Volumes/%s/%s/%s", volume.Database, volume.Schema, volume.Name) + conn.SetProp("internal_volume_path", volumePath) + + return volumePath, nil } // VolumePUT uploads a local file to a Databricks volume using SQL commands @@ -979,7 +987,7 @@ func (conn *DatabricksConn) CopyViaVolume(table Table, df *iop.Dataflow) (count }() // Get volume path for upload - format: /Volumes/catalog/schema/volume/path - volumePrefix := conn.GetProp("internal_volume") + volumePrefix := conn.GetProp("internal_volume_path") volumeFolderPath := fmt.Sprintf("%s/%s/%s", volumePrefix, env.CleanTableName(tableFName), g.NowFileStr()) @@ -1005,7 +1013,7 @@ func (conn *DatabricksConn) CopyViaVolume(table Table, df *iop.Dataflow) (count err = conn.VolumePUT(folderPath, file.Node.Path(), volumeFilePath) if err != nil { - df.Context.CaptureErr(g.Error(err, "Error copying to Databricks Volume: "+conn.GetProp("internal_volume"))) + df.Context.CaptureErr(g.Error(err, "Error copying to Databricks Volume: "+conn.GetProp("internal_volume_path"))) } return volumeFilePath } @@ -1054,7 +1062,7 @@ func (conn *DatabricksConn) UnloadViaVolume(tables ...Table) (filePath string, u } // Get volume path for export - volumePrefix := conn.GetProp("internal_volume") + volumePrefix := conn.GetProp("internal_volume_path") volumeFolderPath := fmt.Sprintf("%s/%s/%s", volumePrefix, tempCloudStorageFolder, g.NowFileStr()) diff --git a/core/dbio/database/database_elasticsearch.go b/core/dbio/database/database_elasticsearch.go index f7508f11f..53b2ebf3d 100644 --- a/core/dbio/database/database_elasticsearch.go +++ b/core/dbio/database/database_elasticsearch.go @@ -145,27 +145,27 @@ func (conn *ElasticsearchConn) GetTableColumns(table *Table, fields ...string) ( } defer mapping.Body.Close() - var mappingResponse map[string]interface{} + var mappingResponse map[string]any if err := json.NewDecoder(mapping.Body).Decode(&mappingResponse); err != nil { return columns, g.Error(err, "could not decode mapping response") } // Navigate to properties - indexMapping, ok := mappingResponse[table.Name].(map[string]interface{}) + indexMapping, ok := mappingResponse[table.Name].(map[string]any) if !ok { - return columns, g.Error("unexpected mapping structure for index %s", table.Name) + return columns, g.Error("unexpected mapping structure for index %s: %s", table.Name, g.Marshal(mappingResponse[table.Name])) } - mappings, ok := indexMapping["mappings"].(map[string]interface{}) + mappings, ok := indexMapping["mappings"].(map[string]any) if !ok { return columns, g.Error("no mappings found for index %s", table.Name) } - properties, ok := mappings["properties"].(map[string]interface{}) + properties, ok := mappings["properties"].(map[string]any) if !ok { // Try ES7+ structure where type is implicit - if props, ok := mappings["_doc"].(map[string]interface{}); ok { - properties = props["properties"].(map[string]interface{}) + if props, ok := mappings["_doc"].(map[string]any); ok { + properties = props["properties"].(map[string]any) } else { // If no properties found, return a single JSON column columns = append(columns, iop.Column{ @@ -182,10 +182,10 @@ func (conn *ElasticsearchConn) GetTableColumns(table *Table, fields ...string) ( } position := 1 - var processProperties func(prefix string, props map[string]interface{}) - processProperties = func(prefix string, props map[string]interface{}) { + var processProperties func(prefix string, props map[string]any) + processProperties = func(prefix string, props map[string]any) { for fieldName, fieldDef := range props { - fieldDefMap, ok := fieldDef.(map[string]interface{}) + fieldDefMap, ok := fieldDef.(map[string]any) if !ok { continue } @@ -194,7 +194,7 @@ func (conn *ElasticsearchConn) GetTableColumns(table *Table, fields ...string) ( // Handle nested objects if fieldType == "object" { - if nestedProps, ok := fieldDefMap["properties"].(map[string]interface{}); ok { + if nestedProps, ok := fieldDefMap["properties"].(map[string]any); ok { newPrefix := prefix if prefix != "" { newPrefix = prefix + "." @@ -284,7 +284,7 @@ func (conn *ElasticsearchConn) BulkExportFlow(table Table) (df *iop.Dataflow, er return df, nil } -func (conn *ElasticsearchConn) StreamRowsContext(ctx context.Context, tableName string, Opts ...map[string]interface{}) (ds *iop.Datastream, err error) { +func (conn *ElasticsearchConn) StreamRowsContext(ctx context.Context, tableName string, Opts ...map[string]any) (ds *iop.Datastream, err error) { opts := getQueryOptions(Opts) Limit := int64(0) // infinite if val := cast.ToInt64(opts["limit"]); val > 0 { @@ -292,14 +292,14 @@ func (conn *ElasticsearchConn) StreamRowsContext(ctx context.Context, tableName } // Handle incremental and backfill options - var searchBody map[string]interface{} + var searchBody map[string]any if updateKey := cast.ToString(opts["update_key"]); updateKey != "" { if incrementalValue := cast.ToString(opts["value"]); incrementalValue != "" { // Incremental mode - searchBody = map[string]interface{}{ - "query": map[string]interface{}{ - "range": map[string]interface{}{ - updateKey: map[string]interface{}{ + searchBody = map[string]any{ + "query": map[string]any{ + "range": map[string]any{ + updateKey: map[string]any{ "gt": incrementalValue, }, }, @@ -308,10 +308,10 @@ func (conn *ElasticsearchConn) StreamRowsContext(ctx context.Context, tableName } else if startValue := cast.ToString(opts["start_value"]); startValue != "" { if endValue := cast.ToString(opts["end_value"]); endValue != "" { // Backfill mode - searchBody = map[string]interface{}{ - "query": map[string]interface{}{ - "range": map[string]interface{}{ - updateKey: map[string]interface{}{ + searchBody = map[string]any{ + "query": map[string]any{ + "range": map[string]any{ + updateKey: map[string]any{ "gte": startValue, "lte": endValue, }, @@ -324,9 +324,9 @@ func (conn *ElasticsearchConn) StreamRowsContext(ctx context.Context, tableName // If no specific query is provided, use match_all if searchBody == nil { - searchBody = map[string]interface{}{ - "query": map[string]interface{}{ - "match_all": map[string]interface{}{}, + searchBody = map[string]any{ + "query": map[string]any{ + "match_all": map[string]any{}, }, } } @@ -358,7 +358,7 @@ func (conn *ElasticsearchConn) StreamRowsContext(ctx context.Context, tableName return nil, g.Error("could not execute search (status %s) => %s", res.StatusCode, string(bytes)) } - var searchResponse map[string]interface{} + var searchResponse map[string]any if err := json.NewDecoder(res.Body).Decode(&searchResponse); err != nil { return nil, g.Error(err, "could not decode search response") } @@ -419,7 +419,7 @@ type elasticDecoder struct { conn *ElasticsearchConn ctx context.Context scrollID string - searchResponse map[string]interface{} + searchResponse map[string]any hits []interface{} currentHit int limit uint64 @@ -464,7 +464,7 @@ func (d *elasticDecoder) Decode(obj interface{}) error { } // Get hits from response - hits, ok := d.searchResponse["hits"].(map[string]interface{}) + hits, ok := d.searchResponse["hits"].(map[string]any) if !ok { return g.Error("hits not found in response") } @@ -484,20 +484,20 @@ func (d *elasticDecoder) Decode(obj interface{}) error { } // Get next hit - hit, ok := d.hits[d.currentHit].(map[string]interface{}) + hit, ok := d.hits[d.currentHit].(map[string]any) if !ok { d.currentHit++ return d.Decode(obj) // skip invalid hit } - source, ok := hit["_source"].(map[string]interface{}) + source, ok := hit["_source"].(map[string]any) if !ok { d.currentHit++ return d.Decode(obj) // skip invalid source } // Set the source as the object to decode - objMap, ok := obj.(*map[string]interface{}) + objMap, ok := obj.(*map[string]any) if !ok { return g.Error("invalid object type for decoding") } @@ -520,7 +520,7 @@ func (conn *ElasticsearchConn) GetSchemas() (data iop.Dataset, err error) { } defer indices.Body.Close() - var indicesResponse []map[string]interface{} + var indicesResponse []map[string]any if err := json.NewDecoder(indices.Body).Decode(&indicesResponse); err != nil { return data, g.Error(err, "could not decode indices response") } @@ -547,7 +547,7 @@ func (conn *ElasticsearchConn) GetTables(schema string) (data iop.Dataset, err e } defer mapping.Body.Close() - var mappingResponse map[string]interface{} + var mappingResponse map[string]any if err := json.NewDecoder(mapping.Body).Decode(&mappingResponse); err != nil { return data, g.Error(err, "could not decode mapping response") } diff --git a/core/dbio/database/database_iceberg.go b/core/dbio/database/database_iceberg.go index 8f0910b07..c094b8c2b 100644 --- a/core/dbio/database/database_iceberg.go +++ b/core/dbio/database/database_iceberg.go @@ -1138,7 +1138,10 @@ func (conn *IcebergConn) DropTable(tableNames ...string) (err error) { // purge table catalog, ok := conn.Catalog.(*rest.Catalog) if ok { - err = catalog.PurgeTable(conn.context.Ctx, identifier) + if err = catalog.PurgeTable(conn.context.Ctx, identifier); err != nil { + // try just drop + err = conn.Catalog.DropTable(conn.context.Ctx, identifier) + } } else { err = conn.Catalog.DropTable(conn.context.Ctx, identifier) } @@ -1606,8 +1609,8 @@ func (conn *IcebergConn) iopTypeToIcebergPrimitiveType(col iop.Column) iceberg.T case iop.FloatType: return iceberg.PrimitiveTypes.Float64 case iop.DecimalType: - precision := lo.Ternary(col.DbPrecision > 0, col.DbPrecision, env.DdlMinDecLength) - scale := lo.Ternary(col.DbScale > 0, col.DbScale, env.DdlMinDecScale) + precision := lo.Ternary(col.DbPrecision > env.DdlMinDecLength, col.DbPrecision, env.DdlMinDecLength) + scale := lo.Ternary(col.DbScale > env.DdlMinDecScale, col.DbScale, env.DdlMinDecScale) return iceberg.DecimalTypeOf(precision, scale) case iop.DateType: return iceberg.PrimitiveTypes.Date diff --git a/core/dbio/database/database_mongo.go b/core/dbio/database/database_mongo.go index 213ead0c3..d1d617311 100644 --- a/core/dbio/database/database_mongo.go +++ b/core/dbio/database/database_mongo.go @@ -153,7 +153,7 @@ func (conn *MongoDBConn) GetTableColumns(table *Table, fields ...string) (column return data.Columns, nil } -func (conn *MongoDBConn) ExecContext(ctx context.Context, sql string, args ...interface{}) (result sql.Result, err error) { +func (conn *MongoDBConn) ExecContext(ctx context.Context, sql string, args ...any) (result sql.Result, err error) { return nil, g.Error("ExecContext not implemented on MongoConn") } @@ -178,7 +178,115 @@ func (conn *MongoDBConn) BulkExportFlow(table Table) (df *iop.Dataflow, err erro return } -func (conn *MongoDBConn) StreamRowsContext(ctx context.Context, collectionName string, Opts ...map[string]interface{}) (ds *iop.Datastream, err error) { +// processMongoFilter recursively processes filter values to convert ObjectID strings +func (conn *MongoDBConn) processMongoFilter(filter any) any { + switch v := filter.(type) { + case map[string]any: + // Process map recursively + result := make(map[string]any) + for key, val := range v { + if key == "_id" || strings.HasSuffix(key, "_id") { + // Special handling for _id fields + result[key] = conn.processObjectIDValue(val) + } else { + result[key] = conn.processMongoFilter(val) + } + } + return result + case map[any]any: + // Process map recursively + result := make(map[string]any) + for key, val := range v { + keyStr := cast.ToString(key) + if keyStr == "_id" || strings.HasSuffix(keyStr, "_id") { + result[keyStr] = conn.processObjectIDValue(val) + } else { + result[keyStr] = conn.processMongoFilter(val) + } + } + return result + case []any: + // Process array recursively + result := make([]any, len(v)) + for i, item := range v { + result[i] = conn.processMongoFilter(item) + } + return result + case string: + // Check if string is ObjectID format + if conn.isObjectIDString(v) { + oid, err := primitive.ObjectIDFromHex(v) + if err == nil { + return oid + } + } + return v + default: + return filter + } +} + +// processObjectIDValue handles ObjectID conversion for _id fields +func (conn *MongoDBConn) processObjectIDValue(val any) any { + switch v := val.(type) { + case string: + // Direct ObjectID hex string + if conn.isObjectIDString(v) { + if oid, err := primitive.ObjectIDFromHex(v); err == nil { + return oid + } + } + // ObjectId("...") format + if strings.HasPrefix(v, "ObjectId(") && strings.HasSuffix(v, ")") { + hex := strings.TrimSuffix(strings.TrimPrefix(v, "ObjectId(\""), "\")") + if oid, err := primitive.ObjectIDFromHex(hex); err == nil { + return oid + } + } + return v + case map[string]any: + // Handle operators like $gte, $lt + result := make(map[string]any) + for op, opVal := range v { + if strings.HasPrefix(op, "$") { + result[op] = conn.processObjectIDValue(opVal) + } else { + result[op] = conn.processMongoFilter(opVal) + } + } + return result + case map[any]any: + // Handle operators like $gte, $lt + result := make(map[string]any) + for op, opVal := range v { + opStr := cast.ToString(op) + if strings.HasPrefix(opStr, "$") { + result[opStr] = conn.processObjectIDValue(opVal) + } else { + result[opStr] = conn.processMongoFilter(opVal) + } + } + return result + default: + return conn.processMongoFilter(val) + } +} + +// isObjectIDString checks if a string is a valid ObjectID hex format +func (conn *MongoDBConn) isObjectIDString(s string) bool { + // ObjectID is 24 hex characters + if len(s) != 24 { + return false + } + for _, c := range s { + if !((c >= '0' && c <= '9') || (c >= 'a' && c <= 'f') || (c >= 'A' && c <= 'F')) { + return false + } + } + return true +} + +func (conn *MongoDBConn) StreamRowsContext(ctx context.Context, collectionName string, Opts ...map[string]any) (ds *iop.Datastream, err error) { opts := getQueryOptions(Opts) Limit := int64(0) // infinite if val := cast.ToInt64(opts["limit"]); val > 0 { @@ -202,7 +310,12 @@ func (conn *MongoDBConn) StreamRowsContext(ctx context.Context, collectionName s filter := bson.D{} if filterOpt, ok := opts["filter"]; ok { - // Convert filter option to bson.D + // Process the filter to convert ObjectID strings + if filterString := g.Marshal(filterOpt); strings.Contains(filterString, "ObjectId(") || strings.Contains(filterString, `"_id"`) { + filterOpt = conn.processMongoFilter(filterOpt) + } + + // Convert processed filter to bson.D switch v := filterOpt.(type) { case map[any]any: // Simple filter format: {"field": "value", "field2": {"$gt": 100}} @@ -353,7 +466,7 @@ func (conn *MongoDBConn) GetSchemas() (data iop.Dataset, err error) { data = iop.NewDataset(iop.NewColumnsFromFields("schema_name")) for _, db := range res.Databases { - data.Append([]interface{}{db.Name}) + data.Append([]any{db.Name}) } return data, nil @@ -370,7 +483,7 @@ func (conn *MongoDBConn) GetTables(schema string) (data iop.Dataset, err error) data = iop.NewDataset(iop.NewColumnsFromFields("table_name")) for _, name := range names { - data.Append([]interface{}{name}) + data.Append([]any{name}) } return data, nil diff --git a/core/dbio/database/database_prometheus.go b/core/dbio/database/database_prometheus.go index 773c98650..08a08d460 100644 --- a/core/dbio/database/database_prometheus.go +++ b/core/dbio/database/database_prometheus.go @@ -113,7 +113,7 @@ func (conn *PrometheusConn) GetSQLColumns(table Table) (columns iop.Columns, err {Name: "timestamp", Type: iop.BigIntType, Position: 4}, {Name: "value", Type: iop.DecimalType, Position: 5}, } - + // Try to get actual columns by querying with a very small time range if table.SQL != "" { // Extract the base query without options @@ -121,7 +121,7 @@ func (conn *PrometheusConn) GetSQLColumns(table Table) (columns iop.Columns, err if idx := strings.Index(baseQuery, "#"); idx != -1 { baseQuery = strings.TrimSpace(baseQuery[:idx]) } - + // Query with a minimal time range to get column structure testOpts := g.M( "start", "now-1m", @@ -130,7 +130,7 @@ func (conn *PrometheusConn) GetSQLColumns(table Table) (columns iop.Columns, err "limit", 1, "get_columns", true, ) - + ds, err := conn.StreamRowsContext(conn.Context().Ctx, baseQuery, testOpts) if err == nil && ds != nil { // Wait for columns to be initialized @@ -141,7 +141,7 @@ func (conn *PrometheusConn) GetSQLColumns(table Table) (columns iop.Columns, err ds.Close() } } - + return columns, nil } @@ -194,6 +194,11 @@ func (conn *PrometheusConn) BulkExportFlow(table Table) (df *iop.Dataflow, err e options["columns"] = table.Columns } + // Pass transforms configuration to datastream + if transformStr := conn.GetProp("transforms"); transformStr != "" { + options["transforms"] = transformStr + } + ds, err := conn.StreamRowsContext(conn.Context().Ctx, table.SQL, options) if err != nil { return df, g.Error(err, "could start datastream") @@ -559,7 +564,28 @@ func (conn *PrometheusConn) StreamRowsContext(ctx context.Context, query string, } } - ds = data.Stream(conn.Props()) + // Create the datastream with connection properties + props := conn.Props() + + // Pass transforms through to the datastream configuration + if transformsOpt := opts["transforms"]; transformsOpt != nil { + props["transforms"] = cast.ToString(transformsOpt) + } + + // Create a temporary datastream to initialize transforms before streaming the data + tempDs := iop.NewDatastream(data.Columns) + tempDs.SetConfig(props) + + // If transforms are configured, apply them manually to the existing data + // since Prometheus bypasses the normal iterator pattern + if transformsOpt := opts["transforms"]; transformsOpt != nil && len(data.Rows) > 0 && !cast.ToBool(opts["get_columns"]) { + err := conn.applyTransformsToData(tempDs, &data) + if err != nil { + return nil, err + } + } + + ds = data.Stream(props) return } @@ -733,7 +759,13 @@ func (conn *PrometheusConn) StreamRowsChunked(queryContext *g.Context, query str // Create the datastream ds = iop.NewDatastreamContext(queryContext.Ctx, iop.Columns{}) - ds.SetConfig(conn.Props()) + + // Set connection properties and transforms + props := conn.Props() + if transformsOpt := opts["transforms"]; transformsOpt != nil { + props["transforms"] = cast.ToString(transformsOpt) + } + ds.SetConfig(props) // Process in chunks go func() { @@ -1013,3 +1045,35 @@ func createPrometheusColumnsFromVector(metricMap map[string]string, sample *mode return columns } + +// applyTransformsToData applies transforms to the collected Prometheus data +func (conn *PrometheusConn) applyTransformsToData(ds *iop.Datastream, data *iop.Dataset) error { + if ds.Sp == nil { + return nil + } + if ds.Sp.Config.Transforms == nil { + return nil + } + + transforms := ds.Sp.Config.Transforms + + // Apply transforms to each row in the collected data + for i, row := range data.Rows { + newRow, err := transforms.Evaluate(row) + if err != nil { + return g.Error(err) + } + data.Rows[i] = newRow + } + + // set columns + data.Columns = ds.Columns + + // set inferred if casting + // so that the datastream doesn't re-infer + if transforms.Casted() { + data.Inferred = true + } + + return nil +} diff --git a/core/dbio/database/database_redshift.go b/core/dbio/database/database_redshift.go index 6b843b6aa..6c86d73f4 100755 --- a/core/dbio/database/database_redshift.go +++ b/core/dbio/database/database_redshift.go @@ -114,6 +114,38 @@ func (conn *RedshiftConn) getS3Props() []string { return s3Props } +func (conn *RedshiftConn) makeCopyCredentialString() (cred string) { + + AwsID := conn.GetProp("AWS_ACCESS_KEY_ID") + AwsAccessKey := conn.GetProp("AWS_SECRET_ACCESS_KEY") + AwsSessionToken := conn.GetProp("AWS_SESSION_TOKEN") + AwsRole := conn.GetProp("AWS_ROLE_ARN") + + template := "" + + if AwsID != "" || AwsSessionToken != "" { + template = template + "\n" + `credentials 'aws_access_key_id={aws_access_key_id};aws_secret_access_key={aws_secret_access_key}` + if AwsSessionToken != "" { + template = template + ";token={aws_session_token}" + } + template += "'" + } + + if AwsRole != "" { + template = template + "\n" + `iam_role '{aws_role_arn}'` + } + + cred = g.R( + template, + "aws_access_key_id", AwsID, + "aws_secret_access_key", AwsAccessKey, + "aws_session_token", AwsSessionToken, + "aws_role_arn", AwsRole, + ) + + return strings.TrimSpace(cred) +} + // Unload unloads a query to S3 func (conn *RedshiftConn) Unload(ctx *g.Context, fileFormat dbio.FileType, tables ...Table) (s3Path string, err error) { @@ -123,19 +155,15 @@ func (conn *RedshiftConn) Unload(ctx *g.Context, fileFormat dbio.FileType, table AwsID := conn.GetProp("AWS_ACCESS_KEY_ID") AwsAccessKey := conn.GetProp("AWS_SECRET_ACCESS_KEY") - AwsSessionToken := conn.GetProp("AWS_SESSION_TOKEN") - - AwsSessionTokenExpr := "" - if AwsSessionToken != "" { - AwsSessionTokenExpr = g.F(";token=%s", AwsSessionToken) - } + AwsRole := conn.GetProp("AWS_ROLE_ARN") + credentialExpr := conn.makeCopyCredentialString() // set format options based on fileformat formatOptions := "" if fileFormat == dbio.FileTypeParquet { formatOptions = g.F("PARQUET") } else { - formatOptions = g.F("GZIP CSV NULL '\\N' HEADER DELIMITER ','") + formatOptions = g.F(`GZIP CSV NULL '\\N' HEADER DELIMITER ','`) } g.Info("unloading from redshift to s3") @@ -158,9 +186,7 @@ func (conn *RedshiftConn) Unload(ctx *g.Context, fileFormat dbio.FileType, table conn.template.Core["copy_to_s3"], "sql", g.F("select * from %s", tempTable.Name), "s3_path", s3PathPart, - "aws_access_key_id", AwsID, - "aws_secret_access_key", AwsAccessKey, - "aws_session_token_expr", AwsSessionTokenExpr, + "credential_expr", credentialExpr, "parallel", conn.GetProp("PARALLEL"), "format_options", formatOptions, ) @@ -182,9 +208,7 @@ func (conn *RedshiftConn) Unload(ctx *g.Context, fileFormat dbio.FileType, table conn.template.Core["copy_to_s3"], "sql", sql, "s3_path", s3PathPart, - "aws_access_key_id", AwsID, - "aws_secret_access_key", AwsAccessKey, - "aws_session_token_expr", AwsSessionTokenExpr, + "credential_expr", credentialExpr, "parallel", conn.GetProp("PARALLEL"), "format_options", formatOptions, ) @@ -193,6 +217,7 @@ func (conn *RedshiftConn) Unload(ctx *g.Context, fileFormat dbio.FileType, table if err != nil { cleanSQL := strings.ReplaceAll(unloadSQL, AwsID, "*****") cleanSQL = strings.ReplaceAll(cleanSQL, AwsAccessKey, "*****") + cleanSQL = strings.ReplaceAll(cleanSQL, AwsRole, "*****") err = g.Error(err, fmt.Sprintf("SQL Error for %s:\n%s", s3PathPart, cleanSQL)) queryContext.CaptureErr(err) } @@ -437,11 +462,7 @@ func (conn *RedshiftConn) CopyFromS3(tableFName, s3Path string, columns iop.Colu err = g.Error("Need to set 'AWS_ACCESS_KEY_ID' and 'AWS_SECRET_ACCESS_KEY' or 'AWS_SESSION_TOKEN' to copy to redshift from S3") return } - - AwsSessionTokenExpr := "" - if AwsSessionToken != "" { - AwsSessionTokenExpr = g.F(";token=%s", AwsSessionToken) - } + credentialExpr := conn.makeCopyCredentialString() tgtColumns := conn.GetType().QuoteNames(columns.Names()...) @@ -452,9 +473,7 @@ func (conn *RedshiftConn) CopyFromS3(tableFName, s3Path string, columns iop.Colu "tgt_table", tableFName, "tgt_columns", strings.Join(tgtColumns, ", "), "s3_path", s3Path, - "aws_access_key_id", AwsID, - "aws_secret_access_key", AwsAccessKey, - "aws_session_token_expr", AwsSessionTokenExpr, + "credential_expr", credentialExpr, ) sql = conn.setEmptyAsNull(sql) diff --git a/core/dbio/database/database_snowflake.go b/core/dbio/database/database_snowflake.go index 515f70eda..99f0797d5 100755 --- a/core/dbio/database/database_snowflake.go +++ b/core/dbio/database/database_snowflake.go @@ -190,22 +190,22 @@ createNew: Name: "SLING_STAGING", Dialect: dbio.TypeDbSnowflake, } - conn.Exec("USE SCHEMA " + defStaging.Schema + noDebugKey) - _, err := conn.Exec("CREATE STAGE IF NOT EXISTS " + defStaging.FullName()) + + _, err := conn.Exec("CREATE STAGE IF NOT EXISTS " + defStaging.FDQN()) if err != nil { - g.Warn("Tried to create Internal Snowflake Stage but failed.\n" + g.ErrMsgSimple(err)) + g.Warn("Tried to create Internal Snowflake Stage but failed. Please provide a stage name with `internal_stage`.\n" + g.ErrMsgSimple(err)) return "" } conn.SetProp("schema", schema) - conn.SetProp("internal_stage", defStaging.FullName()) + conn.SetProp("internal_stage", defStaging.FDQN()) } else { defStaging, _ := ParseTableName(internalStage, dbio.TypeDbSnowflake) - if defStaging.Schema != schema { - // create new staging if schema is different + if defStaging.Schema == "" { + g.Warn("did not specify a fully qualified stage name in `internal_stage`. Need: `DATABASE.SCHEMA.TABLE` format. Using default.") + // create new staging if schema is not provided internalStage = "" goto createNew } - conn.Exec("USE SCHEMA " + defStaging.Schema + noDebugKey) } return conn.GetProp("internal_stage") } @@ -769,17 +769,6 @@ func (conn *SnowflakeConn) CopyViaStage(table Table, df *iop.Dataflow) (count ui // fileFormat = dbio.FileTypeParquet } - // get target columns - columns, err := conn.GetSQLColumns(table) - if err != nil { - return 0, g.Error("could not get columns for %s", table.FullName()) - } - - columns, err = conn.ValidateColumnNames(columns, df.Columns.Names()) - if err != nil { - return 0, g.Error("could not validate columns prior to COPY from STAGE for %s", table.FullName()) - } - tableFName := table.FullName() // Write the ds to a temp file @@ -898,6 +887,21 @@ func (conn *SnowflakeConn) CopyViaStage(table Table, df *iop.Dataflow) (count ui return } + // get target columns + columns, err := conn.GetSQLColumns(table) + if err != nil { + err = g.Error("could not get columns for %s", table.FullName()) + df.Context.CaptureErr(err) + return + } + + columns, err = conn.ValidateColumnNames(columns, df.Columns.Names()) + if err != nil { + err = g.Error("could not validate columns prior to COPY from STAGE for %s", table.FullName()) + df.Context.CaptureErr(err) + return + } + tgtColumns := make([]string, len(columns)) for i, name := range columns.Names() { tgtColumns[i] = conn.Quote(name) diff --git a/core/dbio/database/database_starrocks.go b/core/dbio/database/database_starrocks.go index a8e255e9a..9a36cc14c 100755 --- a/core/dbio/database/database_starrocks.go +++ b/core/dbio/database/database_starrocks.go @@ -508,6 +508,7 @@ func (conn *StarRocksConn) StreamLoad(feURL, tableFName string, df *iop.Dataflow "timeout": "300", "columns": strings.Join(colNames, ", "), + "strict": "true", "format": "JSON", "strip_outer_array": "true", } diff --git a/core/dbio/database/schemata.go b/core/dbio/database/schemata.go index 4a326b54f..b1531738f 100644 --- a/core/dbio/database/schemata.go +++ b/core/dbio/database/schemata.go @@ -380,6 +380,7 @@ func (t *Table) Select(Opts ...SelectOptions) (sql string) { "where_cond", opts.Where, "where_clause", whereClause, "where_and", whereAnd, + "fields", fieldsStr, ) return diff --git a/core/dbio/database/transaction.go b/core/dbio/database/transaction.go index 649ac8ad1..d28feaf77 100644 --- a/core/dbio/database/transaction.go +++ b/core/dbio/database/transaction.go @@ -227,7 +227,7 @@ func (t *BaseTransaction) InsertBatchStream(tableFName string, ds *iop.Datastrea func (t *BaseTransaction) Merge(sourceTable, targetTable string, pkFields []string) (count uint64, err error) { cnt, err := Merge(t.Conn, t, sourceTable, targetTable, pkFields) if err != nil { - err = g.Error(err, "Could not upsert from %s into %s", sourceTable, targetTable) + err = g.Error(err, "Could not merge from %s into %s", sourceTable, targetTable) } count = cast.ToUint64(cnt) return @@ -476,7 +476,7 @@ func Merge(conn Connection, tx Transaction, sourceTable, targetTable string, pkF result, err = conn.ExecMulti(q) } if err != nil { - err = g.Error(err, "Could not upsert") + err = g.Error(err, "Could not merge") return } diff --git a/core/dbio/iop/arrow.go b/core/dbio/iop/arrow.go index 78ac3699a..4218738ef 100644 --- a/core/dbio/iop/arrow.go +++ b/core/dbio/iop/arrow.go @@ -497,8 +497,8 @@ func ColumnsToArrowSchema(columns Columns) *arrow.Schema { case FloatType: arrowType = arrow.PrimitiveTypes.Float64 case DecimalType: - col.DbPrecision = lo.Ternary(col.DbPrecision == 0, int(env.DdlMinDecLength), col.DbPrecision) - col.DbScale = lo.Ternary(col.DbScale == 0, env.DdlMinDecScale, col.DbScale) + col.DbPrecision = lo.Ternary(col.DbPrecision > env.DdlMinDecLength, col.DbPrecision, env.DdlMinDecLength) + col.DbScale = lo.Ternary(col.DbScale > env.DdlMinDecScale, col.DbScale, env.DdlMinDecScale) arrowType = &arrow.Decimal128Type{Precision: int32(col.DbPrecision), Scale: int32(col.DbScale)} case DateType: arrowType = arrow.FixedWidthTypes.Date32 diff --git a/core/dbio/iop/dataflow.go b/core/dbio/iop/dataflow.go index 3e6e6211e..69784a9c2 100644 --- a/core/dbio/iop/dataflow.go +++ b/core/dbio/iop/dataflow.go @@ -113,8 +113,8 @@ func (df *Dataflow) StreamConfig() (cfg StreamConfig) { // SetConfig set the Sp config func (df *Dataflow) SetConfig(cfg StreamConfig) { // don't overwrite transforms if not provided - if cfg.transforms == nil { - cfg.transforms = df.StreamConfig().transforms + if cfg.Transforms == nil { + cfg.Transforms = df.StreamConfig().Transforms } df.mux.Lock() diff --git a/core/dbio/iop/datastream.go b/core/dbio/iop/datastream.go index 0e5e723b5..4f3cb57cd 100644 --- a/core/dbio/iop/datastream.go +++ b/core/dbio/iop/datastream.go @@ -722,11 +722,11 @@ loop: // evaluate transforms, only to determine if type change // will be re-evaluated again below for storing result - if transforms := ds.Sp.Config.transforms; transforms != nil { + if transforms := ds.Sp.Config.Transforms; transforms != nil { // make a copy of row so we don't reference the pointed values rowCopy := make([]any, len(row)) copy(rowCopy, row) - newRow, err := ds.Sp.Config.transforms.Evaluate(rowCopy) + newRow, err := ds.Sp.Config.Transforms.Evaluate(rowCopy) if ds.Context.CaptureErr(err) { break loop } @@ -740,7 +740,7 @@ loop: newType := ds.Sp.CheckType(newVal) if oldType != newType && colType != newType { switch { - case colType.IsDatetime() && TimestampzType.IsDatetime(): // leave as is on orig column + case colType.IsDatetime() && newType.IsDatetime(): // leave as is on orig column case colType.IsDecimal() && newType.IsNumber(): // leave decimal on orig column case colType.IsString() && newType.IsString(): // leave string as is default: @@ -963,8 +963,8 @@ skipBuffer: row = ds.it.Row } else { // evaluate transforms - if transforms := ds.Sp.Config.transforms; transforms != nil { - ds.it.Row, err = ds.Sp.Config.transforms.Evaluate(ds.it.Row) + if transforms := ds.Sp.Config.Transforms; transforms != nil { + ds.it.Row, err = ds.Sp.Config.Transforms.Evaluate(ds.it.Row) if ds.Context.CaptureErr(err) { break loop } diff --git a/core/dbio/iop/datatype.go b/core/dbio/iop/datatype.go index 625fb85d0..e725da49b 100755 --- a/core/dbio/iop/datatype.go +++ b/core/dbio/iop/datatype.go @@ -140,6 +140,10 @@ func (cs *ColumnStats) DuplicatePercent() float64 { } func init() { + SetSampleSize() +} + +func SetSampleSize() { if val := os.Getenv("SAMPLE_SIZE"); val != "" { SampleSize = cast.ToInt(val) // legacy } @@ -1566,6 +1570,10 @@ func (dct *DecimalColumnTyping) Apply(col *Column) (precision, scale int) { dct.MaxScale = lo.Ternary(dct.MaxScale == 0, env.DdlMaxDecScale, dct.MaxScale) dct.MinPrecision = lo.Ternary(dct.MinPrecision == nil, g.Ptr(env.DdlMinDecLength), dct.MinPrecision) dct.MaxPrecision = lo.Ternary(dct.MaxPrecision == 0, env.DdlMaxDecLength, dct.MaxPrecision) + + precisionDelta := lo.Ternary(precision > env.DdlMinDecLength, precision-env.DdlMinDecLength, 0) + scaleDelta := lo.Ternary(scale > env.DdlMinDecScale, scale-env.DdlMinDecScale, 0) + precision = env.DdlMinDecLength + precisionDelta + scaleDelta // safe if scale if present } if dct.MinPrecision != nil && precision < *dct.MinPrecision { diff --git a/core/dbio/iop/datatype_test.go b/core/dbio/iop/datatype_test.go index 439bfd8b5..84f2a2761 100755 --- a/core/dbio/iop/datatype_test.go +++ b/core/dbio/iop/datatype_test.go @@ -324,6 +324,224 @@ func TestParseDecimal(t *testing.T) { assert.Error(t, err) } +func TestColumnTyping(t *testing.T) { + maxStringLength := 1000 + + type testCase struct { + name string + column Column + columnTyping ColumnTyping + + expectedDecimalPrecision int + expectedDecimalScale int + expectedStringLength int + } + + testCases := []testCase{ + // Decimal column typing tests + { + name: "decimal_sourced_precision_scale", + column: Column{Name: "test", Type: DecimalType, DbPrecision: 10, DbScale: 2, Sourced: true}, + columnTyping: ColumnTyping{Decimal: &DecimalColumnTyping{}}, + expectedDecimalPrecision: 10, + expectedDecimalScale: 2, + }, + { + name: "decimal_sourced_precision_scale_2", + column: Column{Name: "test", Type: DecimalType, DbPrecision: 10, DbScale: 2, Sourced: true}, + columnTyping: ColumnTyping{Decimal: &DecimalColumnTyping{}}, + expectedDecimalPrecision: 10, + expectedDecimalScale: 2, + }, + { + name: "decimal_min_precision_scale", + column: Column{Name: "test", Type: DecimalType, DbPrecision: 5, DbScale: 1, Sourced: false}, + columnTyping: ColumnTyping{Decimal: &DecimalColumnTyping{MinPrecision: g.Ptr(10), MinScale: g.Ptr(3)}}, + expectedDecimalPrecision: 24, + expectedDecimalScale: 3, + }, + { + name: "decimal_max_precision_scale", + column: Column{Name: "test", Type: DecimalType, DbPrecision: 50, DbScale: 15, Sourced: false}, + columnTyping: ColumnTyping{Decimal: &DecimalColumnTyping{MaxPrecision: 20, MaxScale: 10}}, + expectedDecimalPrecision: 20, + expectedDecimalScale: 10, + }, + { + name: "decimal_with_stats", + column: Column{Name: "test", Type: DecimalType, Stats: ColumnStats{MaxLen: 8, MaxDecLen: 3}, Sourced: false}, + columnTyping: ColumnTyping{Decimal: &DecimalColumnTyping{}}, + expectedDecimalPrecision: 24, + expectedDecimalScale: 6, + }, + { + name: "decimal_zero_precision_scale", + column: Column{Name: "test", Type: DecimalType, DbPrecision: 0, DbScale: 0, Sourced: false}, + columnTyping: ColumnTyping{Decimal: &DecimalColumnTyping{}}, + expectedDecimalPrecision: 24, + expectedDecimalScale: 6, + }, + { + name: "decimal_delta", + column: Column{Name: "test", Type: DecimalType, DbPrecision: 0, DbScale: 19, Sourced: false}, + columnTyping: ColumnTyping{Decimal: &DecimalColumnTyping{}}, + expectedDecimalPrecision: 38, + expectedDecimalScale: 19, + }, + + // String column typing tests + { + name: "string_basic_length", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 50}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{}}, + expectedStringLength: 50, + }, + { + name: "string_length_factor", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 50}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{LengthFactor: 2}}, + expectedStringLength: 100, + }, + { + name: "string_length_factor_exceeds_max", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 600}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{LengthFactor: 2}}, + expectedStringLength: 1000, // should cap at maxStringLength + }, + { + name: "string_min_length", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 10}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{MinLength: 50}}, + expectedStringLength: 50, + }, + { + name: "string_max_length", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 200}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{MaxLength: 150}}, + expectedStringLength: 200, // original length since MaxLength doesn't override max + }, + { + name: "string_use_max", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 50}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{UseMax: true}}, + expectedStringLength: 1000, // should use maxStringLength + }, + { + name: "string_use_max_with_custom_max", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 50}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{UseMax: true, MaxLength: 2000}}, + expectedStringLength: 2000, // should use custom MaxLength + }, + { + name: "string_min_length_with_factor", + column: Column{Name: "test", Type: StringType, Stats: ColumnStats{MaxLen: 10}}, + columnTyping: ColumnTyping{String: &StringColumnTyping{LengthFactor: 2, MinLength: 50}}, + expectedStringLength: 50, // factor gives 20, but min is 50 + }, + + // Sourced column precision tests + { + name: "string_sourced_precision", + column: Column{Name: "test", Type: StringType, DbPrecision: 100, Sourced: true}, + columnTyping: ColumnTyping{String: &StringColumnTyping{}}, + expectedStringLength: 100, + }, + { + name: "string_sourced_precision_with_factor", + column: Column{Name: "test", Type: StringType, DbPrecision: 50, Sourced: true}, + columnTyping: ColumnTyping{String: &StringColumnTyping{LengthFactor: 2}}, + expectedStringLength: 100, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + if sct := testCase.columnTyping.String; sct != nil { + var length int + if testCase.column.Sourced && testCase.column.DbPrecision > 0 { + length = sct.Apply(testCase.column.DbPrecision, maxStringLength) + } else { + length = sct.Apply(testCase.column.Stats.MaxLen, maxStringLength) + } + assert.Equal(t, testCase.expectedStringLength, length) + + } else if dct := testCase.columnTyping.Decimal; dct != nil { + precision, scale := dct.Apply(&testCase.column) + assert.Equal(t, testCase.expectedDecimalPrecision, precision) + assert.Equal(t, testCase.expectedDecimalScale, scale) + } + }) + } + + // Keep the original hardcoded test for backward compatibility + col := Column{Name: "test", Type: DecimalType, DbPrecision: 10, DbScale: 0, Sourced: true} + ct := ColumnTyping{Decimal: &DecimalColumnTyping{}} + precision, scale := ct.Decimal.Apply(&col) + assert.Equal(t, 10, precision) + assert.Equal(t, 0, scale) +} + +// Additional test for JSON column typing +func TestColumnTypingJSON(t *testing.T) { + t.Run("json_as_text_false", func(t *testing.T) { + col := Column{Name: "test", Type: JsonType} + jct := JsonColumnTyping{AsText: false} + jct.Apply(&col) + assert.Equal(t, JsonType, col.Type) + }) + + t.Run("json_as_text_true", func(t *testing.T) { + col := Column{Name: "test", Type: JsonType} + jct := JsonColumnTyping{AsText: true} + jct.Apply(&col) + assert.Equal(t, TextType, col.Type) + }) +} + +// Test for MaxDecimals method +func TestColumnTypingMaxDecimals(t *testing.T) { + t.Run("nil_column_typing", func(t *testing.T) { + var ct *ColumnTyping + assert.Equal(t, -1, ct.MaxDecimals()) + }) + + t.Run("nil_decimal_typing", func(t *testing.T) { + ct := &ColumnTyping{} + assert.Equal(t, -1, ct.MaxDecimals()) + }) + + t.Run("max_scale_set", func(t *testing.T) { + ct := &ColumnTyping{ + Decimal: &DecimalColumnTyping{MaxScale: 5}, + } + assert.Equal(t, 5, ct.MaxDecimals()) + }) + + t.Run("min_scale_set_no_max", func(t *testing.T) { + ct := &ColumnTyping{ + Decimal: &DecimalColumnTyping{MinScale: g.Ptr(3)}, + } + assert.Equal(t, 3, ct.MaxDecimals()) + }) + + t.Run("both_scales_set", func(t *testing.T) { + ct := &ColumnTyping{ + Decimal: &DecimalColumnTyping{ + MaxScale: 5, + MinScale: g.Ptr(3), + }, + } + assert.Equal(t, 5, ct.MaxDecimals()) // MaxScale takes precedence + }) + + t.Run("no_scales_set", func(t *testing.T) { + ct := &ColumnTyping{ + Decimal: &DecimalColumnTyping{}, + } + assert.Equal(t, -1, ct.MaxDecimals()) + }) +} + func TestDatasetSort(t *testing.T) { columns := NewColumnsFromFields("col1", "col2") data := NewDataset(columns) diff --git a/core/dbio/iop/json.go b/core/dbio/iop/json.go index 69d5919a4..90b83fa34 100644 --- a/core/dbio/iop/json.go +++ b/core/dbio/iop/json.go @@ -210,6 +210,7 @@ func (js *jsonStream) parseRecords(records []map[string]any) { if !ok { col = &Column{ Name: colName, + Type: js.ds.Sp.CheckType(newRec[colName]), Position: len(js.ds.Columns) + len(colsToAdd) + 1, FileURI: cast.ToString(js.ds.Metadata.StreamURL.Value), } diff --git a/core/dbio/iop/stream_processor.go b/core/dbio/iop/stream_processor.go index ceba08624..3b9d35e27 100644 --- a/core/dbio/iop/stream_processor.go +++ b/core/dbio/iop/stream_processor.go @@ -17,6 +17,7 @@ import ( "unicode/utf8" "github.com/flarco/g" + "github.com/prometheus/common/model" "github.com/samber/lo" "github.com/shopspring/decimal" "github.com/slingdata-io/sling-cli/core/dbio" @@ -72,7 +73,7 @@ type StreamConfig struct { DeleteFile bool `json:"delete"` // whether to delete before writing BoolAsInt bool `json:"-"` Columns Columns `json:"columns"` // list of column types. Can be partial list! likely is! - transforms Transform + Transforms Transform Map map[string]string `json:"-"` } @@ -281,7 +282,7 @@ func NewStreamProcessor() *StreamProcessor { func DefaultStreamConfig() StreamConfig { return StreamConfig{ MaxDecimals: -1, - transforms: nil, + Transforms: nil, Map: map[string]string{"delimiter": "-1"}, } } @@ -295,7 +296,7 @@ func LoaderStreamConfig(header bool) StreamConfig { NullAs: `\N`, DatetimeFormat: "auto", MaxDecimals: -1, - transforms: nil, + Transforms: nil, } } @@ -431,7 +432,7 @@ func (sp *StreamProcessor) SetConfig(configMap map[string]string) { func (sp *StreamProcessor) applyTransforms(transformsPayload string) { stageTransforms := []map[string]string{} g.Unmarshal(transformsPayload, &stageTransforms) - sp.Config.transforms = NewTransform(stageTransforms, sp) + sp.Config.Transforms = NewTransform(stageTransforms, sp) } // CastVal casts the type of an interface based on its value @@ -558,7 +559,7 @@ func (sp *StreamProcessor) CheckType(v any) (typ ColumnType) { return BigIntType // Float types - case float32, float64: + case float32, float64, model.Sample, *model.Sample: return DecimalType // Decimal types @@ -683,7 +684,7 @@ func (sp *StreamProcessor) CastVal(i int, val any, col *Column) any { val = sVal isString = true } - case chJSON: // Clickhouse JSON / Variant + case chJSON: // Clickhouse JSON / Variant, has MarshalJSON() sBytes, _ := v.MarshalJSON() sVal = string(sBytes) case string, *string: @@ -1085,13 +1086,6 @@ func (sp *StreamProcessor) CastToStringE(val any) (valString string, err error) valString = string(v) case *string: valString = *v - case chJSON: // Clickhouse JSON / Variant - var sBytes []byte - sBytes, err = v.MarshalJSON() - if err != nil { - return "", g.Error(err, "could not marshal value to JSON: %#v", v) - } - valString = string(sBytes) case *big.Rat: decCount := 12 if sp.Config.MaxDecimals > -1 { @@ -1100,6 +1094,19 @@ func (sp *StreamProcessor) CastToStringE(val any) (valString string, err error) valString = v.FloatString(decCount) case map[string]string, map[string]any, map[any]any, []any, []string: valString = g.Marshal(v) + case model.Sample: + val = val.(float64) + valString, err = cast.ToStringE(val) + if err != nil { + return "", g.Error(err, "could not cast to string: %#v", v) + } + case chJSON: // Clickhouse JSON / Variant or any with MarshalJSON() + var sBytes []byte + sBytes, err = v.MarshalJSON() + if err != nil { + return "", g.Error(err, "could not marshal value to JSON: %#v", v) + } + valString = string(sBytes) default: valString, err = cast.ToStringE(v) if err != nil { diff --git a/core/dbio/iop/transforms.go b/core/dbio/iop/transforms.go index 2d2bf4345..95a5f756b 100644 --- a/core/dbio/iop/transforms.go +++ b/core/dbio/iop/transforms.go @@ -112,6 +112,7 @@ func (tl TransformLegacyList) HasTransform(t TransformLegacy) bool { type Transform interface { Evaluate(row []any) (newRow []any, err error) + Casted() bool } var NewTransform = func(t []map[string]string, _ *StreamProcessor) Transform { diff --git a/core/dbio/iop/transforms_test.go b/core/dbio/iop/transforms_test.go index 9c4b21853..be6e4a611 100755 --- a/core/dbio/iop/transforms_test.go +++ b/core/dbio/iop/transforms_test.go @@ -1398,9 +1398,9 @@ func TestEvaluatorCheckExpression(t *testing.T) { }, } + eval := NewEvaluator(nil) for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - eval := NewEvaluator(nil) err := eval.Check(tt.expression) if tt.expectError { diff --git a/core/dbio/scripts/test.sh b/core/dbio/scripts/test.sh index 3e1d2f0a0..69f890c4d 100644 --- a/core/dbio/scripts/test.sh +++ b/core/dbio/scripts/test.sh @@ -8,7 +8,7 @@ go test -v -run 'TestConnection' cd - cd iop -go test -timeout 5m -v -run 'TestParseDate|TestDetectDelimiter|TestFIX|TestConstraints|TestDuckDb|TestParquetDuckDb|TestIcebergReader|TestDeltaReader|TestPartition|TestExtractPartitionTimeValue|TestGetLowestPartTimeUnit|TestMatchedPartitionMask|TestGeneratePartURIsFromRange|TestDataset|TestValidateNames|TestExcelDateToTime|TestBinaryToHex|TestBinaryToDecimal|TestArrow|TestFunctions|TestQueue|TestEvaluator|TestTransforms' +go test -timeout 5m -v -run 'TestParseDate|TestDetectDelimiter|TestFIX|TestConstraints|TestDuckDb|TestParquetDuckDb|TestIcebergReader|TestDeltaReader|TestPartition|TestExtractPartitionTimeValue|TestGetLowestPartTimeUnit|TestMatchedPartitionMask|TestGeneratePartURIsFromRange|TestDataset|TestValidateNames|TestExcelDateToTime|TestBinaryToHex|TestBinaryToDecimal|TestArrow|TestFunctions|TestQueue|TestEvaluator|TestTransforms|TestColumnTyping' cd - cd database diff --git a/core/dbio/templates/databricks.yaml b/core/dbio/templates/databricks.yaml index c5fe16076..ea47a849d 100644 --- a/core/dbio/templates/databricks.yaml +++ b/core/dbio/templates/databricks.yaml @@ -1,6 +1,7 @@ core: drop_table: drop table if exists {table} drop_view: drop view if exists {view} + create_schema: create schema if not exists {schema} drop_index: "select 'indexes do not apply for databricks'" create_table: | create table {table} ({col_types}) diff --git a/core/dbio/templates/redshift.yaml b/core/dbio/templates/redshift.yaml index 3136d65f7..bffc35914 100755 --- a/core/dbio/templates/redshift.yaml +++ b/core/dbio/templates/redshift.yaml @@ -34,12 +34,12 @@ core: copy_from_s3: | COPY {tgt_table} ({tgt_columns}) from '{s3_path}' - credentials 'aws_access_key_id={aws_access_key_id};aws_secret_access_key={aws_secret_access_key}{aws_session_token_expr}' + {credential_expr} CSV delimiter ',' EMPTYASNULL BLANKSASNULL GZIP IGNOREHEADER 1 DATEFORMAT 'auto' TIMEFORMAT 'auto' copy_to_s3: | unload ('{sql}') to '{s3_path}' - credentials 'aws_access_key_id={aws_access_key_id};aws_secret_access_key={aws_secret_access_key}{aws_session_token_expr}' + {credential_expr} allowoverwrite {format_options} PARALLEL {parallel} alter_columns: | alter table {table} {col_ddl} diff --git a/core/env/env.go b/core/env/env.go index 322ab6466..4695a6f45 100755 --- a/core/env/env.go +++ b/core/env/env.go @@ -194,6 +194,13 @@ func GreenString(text string) string { return g.Colorize(g.ColorGreen, text) } +func YellowString(text string) string { + if NoColor { + return text + } + return g.Colorize(g.ColorYellow, text) +} + func RedString(text string) string { if NoColor { return text diff --git a/core/sling/config.go b/core/sling/config.go index 653125856..b910f4b58 100644 --- a/core/sling/config.go +++ b/core/sling/config.go @@ -149,6 +149,7 @@ func (cfg *Config) SetDefault() { for k, v := range cfg.Env { os.Setenv(k, v) } + iop.SetSampleSize() // default mode if cfg.Mode == "" { diff --git a/core/sling/replication.go b/core/sling/replication.go index f596325cd..7da545a8f 100644 --- a/core/sling/replication.go +++ b/core/sling/replication.go @@ -610,6 +610,7 @@ func (rd *ReplicationConfig) ProcessChunks() (err error) { table, err = database.ParseTableName(stream.config.SQL, sourceConn.Connection.Type) table.SQL = g.R(table.SQL, "incremental_where_cond", "1=1") table.SQL = g.R(table.SQL, "incremental_value", "null") + table.SQL = g.R(table.SQL, "fields", "*") } if err != nil { diff --git a/core/sling/task.go b/core/sling/task.go index 629e76de4..6fe0d05fa 100644 --- a/core/sling/task.go +++ b/core/sling/task.go @@ -584,23 +584,25 @@ func ErrorHelper(err error) (helpString string) { switch { case contains("utf8") || contains("ascii"): - helpString = "Perhaps the 'transforms' source option could help with encodings? Also try `replace_non_printable`. See https://docs.slingdata.io/sling-cli/run/configuration#source" + helpString = "Perhaps the 'encodings' source option could help? See https://docs.slingdata.io/concepts/replication/source-options#supported-encodings. Also try the `replace_non_printable` transform. See https://docs.slingdata.io/concepts/replication/transforms" case contains("failed to verify certificate"): helpString = "Perhaps specifying `encrypt=true` and `TrustServerCertificate=true` properties could help? See https://docs.slingdata.io/connections/database-connections/sqlserver" case contains("ssl is not enabled on the server"): helpString = "Perhaps setting the 'sslmode' option could help? See https://docs.slingdata.io/connections/database-connections/postgres" case contains("invalid input syntax for type") || (contains(" value ") && contains("is not recognized")) || contains("invalid character value") || contains(" exceeds ") || contains(`could not convert`) || contains("provided schema does not match") || contains("Number out of representable range") || contains("Numeric value", " is not recognized") || contains("out of range") || contains("value too long") || contains("converting", "to", "is unsupported") || contains("stl_load_errors"): - helpString = "Perhaps setting a higher 'SAMPLE_SIZE' environment variable could help? This represents the number of records to process in order to infer column types (especially for file sources). The default is 900. Try 2000 or even higher.\nYou can also manually specify the column types with the `columns` source option. See https://docs.slingdata.io/sling-cli/run/configuration#source\nFurthermore, you can try the `target_options.adjust_column_type` setting to allow Sling to automatically alter the column type on the target side." + helpString = "Perhaps setting a higher 'SLING_SAMPLE_SIZE' environment variable could help? This represents the number of records to process in order to infer column types (especially for file sources). The default is 900. Try 2000 or even higher.\nYou can also manually specify the column types with the `columns` input, (see https://docs.slingdata.io/concepts/replication/columns) \nFurthermore, you can try the `target_options.adjust_column_type` setting to allow Sling to automatically alter the column type on the target side." case contains("bcp import"): - helpString = "If facing issues with Microsoft's BCP, try disabling Bulk Loading with `use_bulk=false`. See https://docs.slingdata.io/sling-cli/run/configuration#target" + helpString = "If facing issues with Microsoft's BCP, try disabling Bulk Loading with `use_bulk=false`. See https://docs.slingdata.io/concepts/replication/target-options" + case contains("pq: canceling statement due to statement timeout "): + helpString = "You could try specifying a `statement_timeout` for your Postgres connection. See https://docs.slingdata.io/connections/database-connections/postgres" case contains("[AppendRow]: converting"): - helpString = "Perhaps using the `adjust_column_type: true` target option could help? See https://docs.slingdata.io/sling-cli/run/configuration#target" + helpString = "Perhaps using the `adjust_column_type: true` target option could help? See https://docs.slingdata.io/concepts/replication/target-options" case contains("mkdir", "permission denied"): helpString = "Perhaps setting the SLING_TEMP_DIR environment variable to a writable folder will help." case contains("canceling statement due to conflict with recovery"): helpString = "Perhaps adjusting the `max_standby_archive_delay` and `max_standby_streaming_delay` settings in the source PG Database could help. See https://stackoverflow.com/questions/14592436/postgresql-error-canceling-statement-due-to-conflict-with-recovery" case contains("wrong number of fields"): - helpString = "Perhaps setting the delimiter (source_options.delimiter) would help? See https://docs.slingdata.io/sling-cli/run/configuration#source" + helpString = "Perhaps setting the delimiter (source_options.delimiter) would help? See https://docs.slingdata.io/concepts/replication/source-options" case contains("not implemented makeGoLangScanType"): helpString = "This is related to the Microsoft go-mssqldb driver, which willingly calls a panic for certain column types (such as geometry columns). See https://github.com/microsoft/go-mssqldb/issues/79 and https://github.com/microsoft/go-mssqldb/pull/32. The workaround is to use Custom SQL, and convert the problematic column type into a varchar." case contains("cannot create parquet value") && contains("from go value of type"): diff --git a/core/sling/task_run_read.go b/core/sling/task_run_read.go index a40c85019..82ca2fc55 100644 --- a/core/sling/task_run_read.go +++ b/core/sling/task_run_read.go @@ -35,6 +35,7 @@ func (t *TaskExecution) ReadFromDB(cfg *Config, srcConn database.Connection) (df st.SQL = g.R(st.SQL, "incremental_value", "null") st.SQL = g.R(st.SQL, "start_value", "null") st.SQL = g.R(st.SQL, "end_value", "null") + st.SQL = g.R(st.SQL, "fields", "*") sTable.Columns, err = srcConn.GetSQLColumns(st) if err != nil { @@ -205,7 +206,7 @@ func (t *TaskExecution) ReadFromDB(cfg *Config, srcConn database.Connection) (df // construct select statement for selected fields or where condition if len(selectFields) > 1 || selectFields[0] != "*" || cfg.Source.Where != "" || cfg.Source.Limit() > 0 { - if sTable.SQL != "" && !cfg.SrcConn.Type.IsNoSQL() { + if sTable.SQL != "" && !cfg.SrcConn.Type.IsNoSQL() && !strings.Contains(sTable.SQL, "{fields}") { // If sTable.SQL is already a query (e.g. from incremental template or custom SQL), // it means the field selection (cfg.Source.Select) is assumed to be handled by its construction. selectFields = []string{"*"} diff --git a/go.mod b/go.mod index fd78c007a..51421d486 100644 --- a/go.mod +++ b/go.mod @@ -44,6 +44,7 @@ require ( github.com/integrii/flaggy v1.5.2 github.com/itchyny/timefmt-go v0.1.6 github.com/jedib0t/go-pretty v4.3.0+incompatible + github.com/jinzhu/copier v0.4.0 github.com/jlaffaye/ftp v0.2.0 github.com/jmespath/go-jmespath v0.4.0 github.com/jmoiron/sqlx v1.2.0 @@ -216,7 +217,6 @@ require ( github.com/jcmturner/goidentity/v6 v6.0.1 // indirect github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect github.com/jcmturner/rpc/v2 v2.0.3 // indirect - github.com/jinzhu/copier v0.4.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect github.com/jpillora/backoff v1.0.0 // indirect