feat: add painless script update action

MehmetFiratKomurcu · firat.komurcu · web-flow · commit d8efcb6075f2 · 2025-05-27T13:47:27.000+03:00
* #31 add painless script feat. * #12 add painless script feat. * 12 add painless script feat. * #12 refactor: reorder fields in Action and Script structs for consistency * create go.sum file --------- Co-authored-by: firat.komurcu <firat.komurcu@trendyol.com>
diff --git a/elasticsearch/action.go b/elasticsearch/action.go
@@ -1,20 +1,28 @@
 package elasticsearch
 
+import "encoding/json"
+
 type ActionType string
 
 const (
-	Index  ActionType = "Index"
-	Delete ActionType = "Delete"
+	Index        ActionType = "Index"
+	Delete       ActionType = "Delete"
+	ScriptUpdate ActionType = "ScriptUpdate"
 )
 
 type Action struct {
 	Routing   *string
-	Type      ActionType
 	IndexName string
+	Type      ActionType
 	Source    []byte
 	ID        []byte
 }
 
+type Script struct {
+	Params map[string]interface{} `json:"params,omitempty"`
+	Source string                 `json:"source"`
+}
+
 func NewDeleteAction(key []byte, routing *string) Action {
 	return Action{
 		ID:      key,
@@ -31,3 +39,13 @@ func NewIndexAction(key []byte, source []byte, routing *string) Action {
 		Type:    Index,
 	}
 }
+
+func NewScriptUpdateAction(id []byte, script Script, routing *string) Action {
+	scriptBytes, _ := json.Marshal(script)
+	return Action{
+		ID:      id,
+		Type:    ScriptUpdate,
+		Source:  scriptBytes,
+		Routing: routing,
+	}
+}
diff --git a/elasticsearch/bulk/bulk.go b/elasticsearch/bulk/bulk.go
@@ -171,10 +171,13 @@ func (b *Bulk) AddActions(
 var (
 	indexPrefix   = []byte(`{"index":{"_index":"`)
 	deletePrefix  = []byte(`{"delete":{"_index":"`)
+	updatePrefix  = []byte(`{"update":{"_index":"`)
+	scriptPrefix  = []byte(`{"script":`)
 	idPrefix      = []byte(`","_id":"`)
 	typePrefix    = []byte(`","_type":"`)
 	routingPrefix = []byte(`","routing":"`)
 	postFix       = []byte(`"}}`)
+	scriptPostfix = []byte(`,"scripted_upsert":true}`)
 )
 
 var metaPool = sync.Pool{
@@ -200,11 +203,15 @@ func isTypeSupported(version string) bool {
 func getEsActionJSON(docID []byte, action elasticsearch2.ActionType, indexName string, routing *string, source []byte, typeName []byte, esVersion string) []byte {
 	meta := metaPool.Get().([]byte)[:0]
 
-	if action == elasticsearch2.Index {
+	switch action {
+	case elasticsearch2.Index:
 		meta = append(meta, indexPrefix...)
-	} else {
+	case elasticsearch2.Delete:
 		meta = append(meta, deletePrefix...)
+	case elasticsearch2.ScriptUpdate:
+		meta = append(meta, updatePrefix...)
 	}
+
 	meta = append(meta, []byte(indexName)...)
 	meta = append(meta, idPrefix...)
 	meta = append(meta, bytes.EscapePredefinedBytes(docID)...)
@@ -217,9 +224,16 @@ func getEsActionJSON(docID []byte, action elasticsearch2.ActionType, indexName s
 		meta = append(meta, typeName...)
 	}
 	meta = append(meta, postFix...)
-	if action == elasticsearch2.Index {
+
+	switch action {
+	case elasticsearch2.Index:
+		meta = append(meta, '\n')
+		meta = append(meta, source...)
+	case elasticsearch2.ScriptUpdate:
 		meta = append(meta, '\n')
+		meta = append(meta, scriptPrefix...)
 		meta = append(meta, source...)
+		meta = append(meta, scriptPostfix...)
 	}
 	meta = append(meta, '\n')
 	return meta
diff --git a/example/script-update/README.md b/example/script-update/README.md
@@ -0,0 +1,120 @@
+# Script Update Example
+
+This example demonstrates how to use painless scripts and partial updates with go-pq-cdc-elasticsearch.
+
+## Features Demonstrated
+
+1. Script Updates
+   - Simple field updates
+   - Conditional updates
+   - Using script parameters
+
+2. Doc Updates
+   - Partial document updates
+   - Field-specific updates
+
+## Prerequisites
+
+- Docker and Docker Compose
+- Go 1.20 or later
+
+## Setup
+
+1. Start the required services:
+
+```bash
+export STACK_VERSION=7.17.11
+docker-compose up -d
+```
+
+2. Wait for all services to be healthy (check with `docker-compose ps`)
+
+3. Create the database table and insert sample data:
+
+```sql
+psql "postgres://script_cdc_user:script_cdc_pass@127.0.0.1/script_cdc_db?replication=database"
+
+CREATE TABLE products (
+    id serial PRIMARY KEY,
+    name text NOT NULL,
+    price decimal(10,2),
+    stock integer,
+    last_updated timestamptz
+);
+
+-- Insert sample data
+INSERT INTO products (name, price, stock, last_updated)
+VALUES 
+    ('Product 1', 99.99, 100, NOW()),
+    ('Product 2', 149.99, 50, NOW()),
+    ('Product 3', 199.99, 25, NOW());
+```
+
+## Running the Example
+
+1. Build and run the example:
+
+```bash
+go mod tidy
+go run main.go
+```
+
+2. Test different update scenarios:
+
+```sql
+-- Update stock using script
+UPDATE products SET stock = stock + 10 WHERE id = 1;
+
+-- Update price conditionally
+UPDATE products SET price = 89.99 WHERE id = 2;
+
+-- Update multiple fields
+UPDATE products SET stock = stock - 5, last_updated = NOW() WHERE id = 3;
+```
+
+## Understanding the Code
+
+The example demonstrates three types of updates:
+
+1. **Script Updates for Stock Changes**:
+
+```go
+script := map[string]interface{}{
+    "source": "ctx._source.stock = params.new_stock",
+    "params": map[string]interface{}{
+        "new_stock": newStock,
+    },
+}
+```
+
+2. **Conditional Script Updates for Price Changes**:
+
+```go
+script := map[string]interface{}{
+    "source": "if (ctx._source.price != params.new_price) { ctx._source.price = params.new_price }",
+    "params": map[string]interface{}{
+        "new_price": newPrice,
+    },
+}
+```
+
+3. **Doc Updates for Other Fields**:
+
+```go
+updateData := map[string]interface{}{
+    "name":         msg.NewData["name"],
+    "last_updated": msg.NewData["last_updated"],
+}
+```
+
+## Monitoring
+
+- Check Elasticsearch indices: http://localhost:9200/products/_search
+- View logs: `docker-compose logs -f`
+- Monitor metrics: http://localhost:8081/metrics
+
+## Cleanup
+
+```bash
+docker-compose down -v
+``` 
diff --git a/example/script-update/docker-compose.yml b/example/script-update/docker-compose.yml
@@ -0,0 +1,83 @@
+version: "3"
+services:
+  postgres:
+    image: postgres:16.2
+    restart: always
+    command: ["-c", "wal_level=logical", "-c", "max_wal_senders=10", "-c", "max_replication_slots=10"]
+    environment:
+      POSTGRES_USER: "script_cdc_user"
+      POSTGRES_PASSWORD: "script_cdc_pass"
+      POSTGRES_DB: "script_cdc_db"
+      POSTGRES_HOST_AUTH_METHOD: trust
+    ports:
+      - 5432:5432
+
+  es01:
+    image: docker.elastic.co/elasticsearch/elasticsearch:7.17.11
+    labels:
+      co.elastic.logs/module: elasticsearch
+    volumes:
+      - es01_data:/usr/share/elasticsearch/data
+    networks:
+      - cdc_net
+    ports:
+      - 9200:9200
+    environment:
+      - node.name=es01
+      - cluster.name=script_cdc_cluster
+      - discovery.type=single-node
+      - ELASTIC_PASSWORD=script_cdc_es_pass
+      - bootstrap.memory_lock=true
+      - xpack.security.enabled=true
+      - xpack.security.http.ssl.enabled=false
+      - xpack.security.transport.ssl.enabled=false
+    ulimits:
+      memlock:
+        soft: -1
+        hard: -1
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "curl -u elastic:script_cdc_es_pass -I -o /dev/null http://localhost:9200 -w '%{http_code}' | grep -q '200'",
+        ]
+      interval: 10s
+      timeout: 10s
+      retries: 120
+
+  kibana:
+    depends_on:
+      es01:
+        condition: service_healthy
+    image: docker.elastic.co/kibana/kibana:7.17.11
+    networks:
+      - cdc_net
+    labels:
+      co.elastic.logs/module: kibana
+    volumes:
+      - kibana1_data:/usr/share/kibana/data
+    ports:
+      - "5601:5601"
+    environment:
+      - SERVERNAME=kibana
+      - ELASTICSEARCH_HOSTS=http://es01:9200
+      - ELASTICSEARCH_USERNAME=elastic
+      - ELASTICSEARCH_PASSWORD=script_cdc_es_pass
+    healthcheck:
+      test:
+        [
+          "CMD-SHELL",
+          "curl -s -I http://localhost:5601 | grep -q 'HTTP/1.1 302 Found'",
+        ]
+      interval: 10s
+      timeout: 10s
+      retries: 120
+
+volumes:
+  postgres_data: null
+  kibana1_data: null
+  es01_data: null
+
+networks:
+  cdc_net:
+    driver: bridge 
diff --git a/example/script-update/go.mod b/example/script-update/go.mod
@@ -0,0 +1,42 @@
+module script-update
+
+go 1.22.5
+
+toolchain go1.23.4
+
+require (
+	github.com/Trendyol/go-pq-cdc v0.0.12
+	github.com/Trendyol/go-pq-cdc-elasticsearch v0.0.0
+)
+
+require (
+	github.com/andybalholm/brotli v1.1.0 // indirect
+	github.com/avast/retry-go/v4 v4.6.0 // indirect
+	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/elastic/go-elasticsearch/v7 v7.17.10 // indirect
+	github.com/go-playground/errors v3.3.0+incompatible // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
+	github.com/jackc/pgx/v5 v5.6.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.17.9 // indirect
+	github.com/lib/pq v1.10.9 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/prometheus/client_golang v1.19.1 // indirect
+	github.com/prometheus/client_model v0.6.1 // indirect
+	github.com/prometheus/common v0.55.0 // indirect
+	github.com/prometheus/procfs v0.15.1 // indirect
+	github.com/valyala/bytebufferpool v1.0.0 // indirect
+	github.com/valyala/fasthttp v1.55.0 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/sync v0.7.0 // indirect
+	golang.org/x/sys v0.21.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
+	google.golang.org/protobuf v1.34.2 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+)
+
+replace github.com/Trendyol/go-pq-cdc-elasticsearch => ../../
diff --git a/example/script-update/go.sum b/example/script-update/go.sum
diff --git a/example/script-update/main.go b/example/script-update/main.go
