Merge pull request #227 from MaterializeInc/format-json

Support FORMAT JSON for sources
MaterializeInc · Jul 7, 2023 · afd70f1 · afd70f1
2 parents 9d0e684 + 8ed69e2
commit afd70f1
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 0 deletions.
diff --git a/pkg/materialize/format_specs.go b/pkg/materialize/format_specs.go
@@ -23,6 +23,7 @@ type SourceFormatSpecStruct struct {
 	Csv      *CsvFormatSpec
 	Bytes    bool
 	Text     bool
+	Json     bool
 }
 
 type SinkAvroFormatSpec struct {
@@ -88,6 +89,9 @@ func GetFormatSpecStruc(v interface{}) SourceFormatSpecStruct {
 	if v, ok := u["text"]; ok {
 		format.Text = v.(bool)
 	}
+	if v, ok := u["json"]; ok {
+		format.Json = v.(bool)
+	}
 	return format
 }
 

diff --git a/pkg/materialize/source_kafka.go b/pkg/materialize/source_kafka.go
@@ -202,6 +202,10 @@ func (b *SourceKafkaBuilder) Create() error {
 		q.WriteString(` FORMAT TEXT`)
 	}
 
+	if b.format.Json {
+		q.WriteString(` FORMAT JSON`)
+	}
+
 	if b.keyFormat.Avro != nil {
 		if b.keyFormat.Avro.SchemaRegistryConnection.Name != "" {
 			q.WriteString(fmt.Sprintf(` KEY FORMAT AVRO USING CONFLUENT SCHEMA REGISTRY CONNECTION %s`, QualifiedName(b.keyFormat.Avro.SchemaRegistryConnection.DatabaseName, b.keyFormat.Avro.SchemaRegistryConnection.SchemaName, b.keyFormat.Avro.SchemaRegistryConnection.Name)))