Add support for [column] ? 'key' operator, fix Attio pagination

Author: exAspArk

README.md

@@ -316,6 +316,7 @@ SELECT * FROM [TABLE] WHERE [JSON_COLUMN]->>'[JSON_KEY]' = '[JSON_VALUE]';
   - [x] TablePlus
   - [x] DBeaver
   - [x] pgAdmin
+  - [x] Grafana
   - [ ] Jupyter notebooks ([#27](https://github.com/BemiHQ/BemiDB/issues/27))
 - [x] Data syncing from other sources
   - [x] Amplitude (incremental)

src/common/iceberg_table_writer.go

@@ -675,7 +675,7 @@ func (writer *IcebergTableWriter) jsonToDuckdbRowValues(rowValues map[string]int
 	}
 	tableColumnNames := make([]string, len(writer.IcebergSchemaColumns))
 	for i, icebergSchemaColumn := range writer.IcebergSchemaColumns {
-		tableColumnNames[i] = strings.ToLower(icebergSchemaColumn.ColumnName) // Debezium JSON keys are lowercase
+		tableColumnNames[i] = icebergSchemaColumn.ColumnName
 	}
 	if len(rowColumnNames) != len(tableColumnNames) {
 		Panic(writer.Config, "Row column names count doesn't match table column names count: "+strings.Join(rowColumnNames, ", ")+" (row) vs "+strings.Join(tableColumnNames, ", ")+" (table)")
@@ -691,7 +691,7 @@ func (writer *IcebergTableWriter) jsonToDuckdbRowValues(rowValues map[string]int
 	// Convert row values to DuckDB values
 	duckdbRowValues := make([]driver.Value, len(writer.IcebergSchemaColumns))
 	for i, icebergSchemaColumn := range writer.IcebergSchemaColumns {
-		columnName := strings.ToLower(icebergSchemaColumn.ColumnName) // Debezium JSON keys are lowercase
+		columnName := icebergSchemaColumn.ColumnName
 		value := rowValues[columnName]
 		duckdbRowValues[i] = icebergSchemaColumn.DuckdbValueFromJson(value)
 	}

src/server/parser_a_expr.go

@@ -62,6 +62,20 @@ func (parser *ParserAExpr) RemappedJsonExtract(node *pgQuery.Node) *pgQuery.Node
 	)
 }
 
+// [column] ? 'key' -> json_exists([column], 'key')
+func (parser *ParserAExpr) RemappedJsonExists(node *pgQuery.Node) *pgQuery.Node {
+	aExpr := parser.AExpr(node)
+	if aExpr == nil || parser.OperatorName(aExpr) != "?" {
+		return node
+	}
+
+	return pgQuery.MakeFuncCallNode(
+		[]*pgQuery.Node{pgQuery.MakeStrNode("json_exists")},
+		[]*pgQuery.Node{aExpr.Lexpr, aExpr.Rexpr},
+		0,
+	)
+}
+
 func (parser *ParserAExpr) OperatorName(aExpr *pgQuery.A_Expr) string {
 	if aExpr.Kind != pgQuery.A_Expr_Kind_AEXPR_OP || len(aExpr.Name) != 1 {
 		return ""

src/server/query_handler_test.go

@@ -931,6 +931,11 @@ func TestHandleQuery(t *testing.T) {
 				"types":       {uint32ToString(pgtype.TextOID)},
 				"values":      {"value"},
 			},
+			"SELECT json_column ? 'key' AS exists FROM postgres.test_table WHERE id = 1": {
+				"description": {"exists"},
+				"types":       {uint32ToString(pgtype.BoolOID)},
+				"values":      {"t"},
+			},
 			"SELECT jsonb_column FROM postgres.test_table WHERE bool_column = FALSE": {
 				"description": {"jsonb_column"},
 				"types":       {uint32ToString(pgtype.TextOID)},

src/server/query_remapper.go

@@ -372,7 +372,7 @@ func (remapper *QueryRemapper) remappedExpressions(node *pgQuery.Node, remappedC
 		remapper.remapSelectStatement(subSelect, permissions, indentLevel+1) // recursion
 	}
 
-	// Comparison
+	// Operator: =, ?, etc.
 	aExpr := node.GetAExpr()
 	if aExpr != nil {
 		node = remapper.remapperExpression.RemappedExpression(node)

src/server/query_remapper_expression.go

@@ -25,7 +25,7 @@ func NewQueryRemapperExpression(config *Config) *QueryRemapperExpression {
 
 func (remapper *QueryRemapperExpression) RemappedExpression(node *pgQuery.Node) *pgQuery.Node {
 	node = remapper.remappedTypeCast(node)
-	node = remapper.remappedArithmeticExpression(node)
+	node = remapper.remappedOperatorExpression(node)
 	node = remapper.remappedCollateClause(node)
 	node = remapper.remappedNullColumnExpression(node)
 	remapper.remapColumnReference(node)
@@ -80,7 +80,7 @@ func (remapper *QueryRemapperExpression) remappedTypeCast(node *pgQuery.Node) *p
 	return node
 }
 
-func (remapper *QueryRemapperExpression) remappedArithmeticExpression(node *pgQuery.Node) *pgQuery.Node {
+func (remapper *QueryRemapperExpression) remappedOperatorExpression(node *pgQuery.Node) *pgQuery.Node {
 	aExpr := remapper.parserAExpr.AExpr(node)
 	if aExpr == nil {
 		return node
@@ -98,6 +98,9 @@ func (remapper *QueryRemapperExpression) remappedArithmeticExpression(node *pgQu
 	// [column]->'value' -> json_extract([column], 'value')
 	node = remapper.parserAExpr.RemappedJsonExtract(node)
 
+	// [column] ? 'key' -> json_exists([column], 'key')
+	node = remapper.parserAExpr.RemappedJsonExists(node)
+
 	return node
 }
 

src/syncer-attio/lib/attio.go

@@ -1,6 +1,7 @@
 package attio
 
 import (
+	"bytes"
 	"encoding/json"
 	"fmt"
 	"io"
@@ -40,23 +41,24 @@ type ListRecordsResponse struct {
 func (attio *Attio) Load(object string, jsonQueueWriter *common.JsonQueueWriter) error {
 	offset := 0
 	for {
-		req, err := http.NewRequest("POST", ATTIO_API_URL+"/objects/"+object+"/records/query", nil)
+		jsonBody, err := json.Marshal(map[string]interface{}{"limit": ATTIO_API_LIMIT, "offset": offset})
+		if err != nil {
+			return err
+		}
+		req, err := http.NewRequest("POST", ATTIO_API_URL+"/objects/"+object+"/records/query", bytes.NewBuffer(jsonBody))
 		if err != nil {
 			return err
 		}
-		q := req.URL.Query()
-		q.Add("limit", common.IntToString(ATTIO_API_LIMIT))
-		q.Add("offset", common.IntToString(offset))
-		req.URL.RawQuery = q.Encode()
-
 		req.Header.Set("Authorization", "Bearer "+attio.Config.ApiAccessToken)
 		req.Header.Set("Content-Type", "application/json")
 
+		common.LogInfo(attio.Config.CommonConfig, "Sending request to Attio:", req.URL.String(), "with body:", string(jsonBody))
 		resp, err := attio.HttpClient.Do(req)
 		if err != nil {
 			return err
 		}
 
+		common.LogDebug(attio.Config.CommonConfig, "Received response from Attio:", resp.Status)
 		if resp.StatusCode != http.StatusOK {
 			body, _ := io.ReadAll(resp.Body)
 			resp.Body.Close()