diff --git a/pf/schema_sidecar.go b/pf/schema_sidecar.go new file mode 100644 index 0000000..e3aa556 --- /dev/null +++ b/pf/schema_sidecar.go @@ -0,0 +1,124 @@ +// +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +package pf + +import ( + "encoding/json" + "os" + "path/filepath" + "strings" +) + +const ( + // SinkSchemaSidecarFileName is the schema sidecar file consumed by the generic runtime. + SinkSchemaSidecarFileName = "sink.schema.json" + + // SchemaTypeNone disables the sink schema sidecar. + SchemaTypeNone = "none" + // SchemaTypeBytes disables the sink schema sidecar. + SchemaTypeBytes = "bytes" + SchemaTypeJSON = "json" + SchemaTypeAvro = "avro" + SchemaTypeString = "string" + SchemaTypeBool = "boolean" + SchemaTypeInt8 = "int8" + SchemaTypeInt16 = "int16" + SchemaTypeInt32 = "int32" + SchemaTypeInt64 = "int64" + SchemaTypeFloat = "float" + SchemaTypeDouble = "double" + SchemaTypeDate = "date" + SchemaTypeTime = "time" + SchemaTypeTimestamp = "timestamp" +) + +// SinkSchema describes the sink schema definition written to the runtime sidecar file. +// SchemaType should use one of the SchemaType* constants for JSON, Avro, or +// primitive schemas. Empty, none, and bytes schema types do not need sidecars. +type SinkSchema struct { + SchemaType string + Name string + SchemaData string + Properties map[string]string +} + +type sinkSchemaSidecarPayload struct { + SchemaType string `json:"schemaType"` + Name string `json:"name"` + SchemaData string `json:"schemaData"` + Properties map[string]string `json:"properties"` +} + +// StartWithSinkSchema writes the sink schema sidecar before starting the function. +func StartWithSinkSchema(funcName interface{}, schema SinkSchema) { + if _, err := WriteSinkSchemaSidecar(schema); err != nil { + panic(err) + } + Start(funcName) +} + +// WriteSinkSchemaSidecar writes sink.schema.json next to the current executable. +func WriteSinkSchemaSidecar(schema SinkSchema) (string, error) { + executable, err := os.Executable() + if err != nil { + return "", err + } + return writeSinkSchemaSidecarToDir(filepath.Dir(executable), schema) +} + +func writeSinkSchemaSidecarToDir(workDir string, schema SinkSchema) (string, error) { + schemaType := strings.TrimSpace(schema.SchemaType) + sidecarPath := filepath.Join(workDir, SinkSchemaSidecarFileName) + if isNoSchemaType(schemaType) { + if err := os.Remove(sidecarPath); err != nil && !os.IsNotExist(err) { + return "", err + } + return "", nil + } + + properties := schema.Properties + if properties == nil { + properties = map[string]string{} + } + payload := sinkSchemaSidecarPayload{ + SchemaType: schemaType, + Name: schema.Name, + SchemaData: schema.SchemaData, + Properties: properties, + } + content, err := json.Marshal(payload) + if err != nil { + return "", err + } + + if err := os.WriteFile(sidecarPath, content, 0644); err != nil { + return "", err + } + return sidecarPath, nil +} + +func isNoSchemaType(schemaType string) bool { + switch strings.ToLower(strings.TrimSpace(schemaType)) { + case "", SchemaTypeBytes, SchemaTypeNone: + return true + default: + return false + } +} diff --git a/pf/schema_sidecar_test.go b/pf/schema_sidecar_test.go new file mode 100644 index 0000000..44cd3e9 --- /dev/null +++ b/pf/schema_sidecar_test.go @@ -0,0 +1,169 @@ +// +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// + +package pf + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" +) + +func TestWriteSinkSchemaSidecarToDirWritesCommonSidecar(t *testing.T) { + workDir := t.TempDir() + schema := SinkSchema{ + SchemaType: SchemaTypeJSON, + Name: "example.Student", + SchemaData: `{"type":"record","name":"Student","fields":[]}`, + Properties: map[string]string{ + "custom": "value", + }, + } + + sidecarPath, err := writeSinkSchemaSidecarToDir(workDir, schema) + if err != nil { + t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err) + } + + if sidecarPath != filepath.Join(workDir, SinkSchemaSidecarFileName) { + t.Fatalf("sidecarPath = %q, want %q", sidecarPath, filepath.Join(workDir, SinkSchemaSidecarFileName)) + } + + content, err := os.ReadFile(sidecarPath) + if err != nil { + t.Fatalf("read sidecar: %v", err) + } + var payload map[string]interface{} + if err := json.Unmarshal(content, &payload); err != nil { + t.Fatalf("unmarshal sidecar: %v", err) + } + + if payload["schemaType"] != "json" { + t.Fatalf("schemaType = %v, want json", payload["schemaType"]) + } + if payload["name"] != "example.Student" { + t.Fatalf("name = %v, want example.Student", payload["name"]) + } + if payload["schemaData"] != `{"type":"record","name":"Student","fields":[]}` { + t.Fatalf("schemaData = %v, want record schema", payload["schemaData"]) + } + properties, ok := payload["properties"].(map[string]interface{}) + if !ok { + t.Fatalf("properties = %T, want object", payload["properties"]) + } + if properties["custom"] != "value" { + t.Fatalf("properties[custom] = %v, want value", properties["custom"]) + } +} + +func TestWriteSinkSchemaSidecarToDirSkipsNoSchemaTypesAndRemovesStaleSidecar(t *testing.T) { + tests := []struct { + name string + schemaType string + }{ + {name: "empty", schemaType: ""}, + {name: "bytes", schemaType: SchemaTypeBytes}, + {name: "uppercase bytes", schemaType: "BYTES"}, + {name: "none", schemaType: SchemaTypeNone}, + {name: "uppercase none", schemaType: "NONE"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + workDir := t.TempDir() + sidecarFile := filepath.Join(workDir, SinkSchemaSidecarFileName) + if err := os.WriteFile(sidecarFile, []byte("stale"), 0644); err != nil { + t.Fatalf("write stale sidecar: %v", err) + } + + sidecarPath, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{SchemaType: tt.schemaType}) + if err != nil { + t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err) + } + if sidecarPath != "" { + t.Fatalf("sidecarPath = %q, want empty", sidecarPath) + } + if _, err := os.Stat(sidecarFile); !os.IsNotExist(err) { + t.Fatalf("sidecar stat error = %v, want not exist", err) + } + }) + } +} + +func TestWriteSinkSchemaSidecarToDirTrimsSchemaTypeAndPreservesCasing(t *testing.T) { + workDir := t.TempDir() + + sidecarPath, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{ + SchemaType: " JSON ", + SchemaData: `{"type":"record","name":"Student","fields":[]}`, + }) + if err != nil { + t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err) + } + + content, err := os.ReadFile(sidecarPath) + if err != nil { + t.Fatalf("read sidecar: %v", err) + } + var payload struct { + SchemaType string `json:"schemaType"` + } + if err := json.Unmarshal(content, &payload); err != nil { + t.Fatalf("unmarshal sidecar: %v", err) + } + if payload.SchemaType != "JSON" { + t.Fatalf("schemaType = %q, want JSON", payload.SchemaType) + } +} + +func TestSchemaTypeBoolUsesPulsarBooleanName(t *testing.T) { + if SchemaTypeBool != "boolean" { + t.Fatalf("SchemaTypeBool = %q, want boolean", SchemaTypeBool) + } +} + +func TestWriteSinkSchemaSidecarToDirWritesEmptyPropertiesObject(t *testing.T) { + workDir := t.TempDir() + + sidecarPath, err := writeSinkSchemaSidecarToDir(workDir, SinkSchema{ + SchemaType: SchemaTypeString, + SchemaData: "", + }) + if err != nil { + t.Fatalf("writeSinkSchemaSidecarToDir returned error: %v", err) + } + + content, err := os.ReadFile(sidecarPath) + if err != nil { + t.Fatalf("read sidecar: %v", err) + } + var payload struct { + Properties map[string]string `json:"properties"` + } + if err := json.Unmarshal(content, &payload); err != nil { + t.Fatalf("unmarshal sidecar: %v", err) + } + if payload.Properties == nil { + t.Fatal("properties is nil, want empty object") + } + if len(payload.Properties) != 0 { + t.Fatalf("properties length = %d, want 0", len(payload.Properties)) + } +}