TRANSFER-467: Raw 2 Table parser (#105)

Author: laskoviymishka

docs/resources/transfer_endpoint.md

@@ -434,6 +434,7 @@ Optional:
 
 - `blank` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kafka_source--parser--blank))
 - `json` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kafka_source--parser--json))
+- `raw_table` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kafka_source--parser--raw_table))
 - `schema_registry` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kafka_source--parser--schema_registry))
 - `tskv` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kafka_source--parser--tskv))
 
@@ -488,6 +489,18 @@ Optional:
 
 
 
+<a id="nestedblock--settings--kafka_source--parser--raw_table"></a>
+### Nested Schema for `settings.kafka_source.parser.raw_table`
+
+Optional:
+
+- `add_headers` (Boolean) Add headers column to output virtual table
+- `add_key` (Boolean) Add key column to output virtual table
+- `add_timestamp` (Boolean) Add timestamp column to output virtual table
+- `keys_as_bytes` (Boolean) Make keys column as `bytes`, for non-utf8 characters
+- `value_as_bytes` (Boolean) Make value column as `bytes`, for non-utf8 characters
+
+
 <a id="nestedblock--settings--kafka_source--parser--schema_registry"></a>
 ### Nested Schema for `settings.kafka_source.parser.schema_registry`
 
@@ -718,6 +731,7 @@ Optional:
 
 - `blank` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kinesis_source--parser--blank))
 - `json` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kinesis_source--parser--json))
+- `raw_table` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kinesis_source--parser--raw_table))
 - `schema_registry` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kinesis_source--parser--schema_registry))
 - `tskv` (Block, Optional) (see [below for nested schema](#nestedblock--settings--kinesis_source--parser--tskv))
 
@@ -772,6 +786,18 @@ Optional:
 
 
 
+<a id="nestedblock--settings--kinesis_source--parser--raw_table"></a>
+### Nested Schema for `settings.kinesis_source.parser.raw_table`
+
+Optional:
+
+- `add_headers` (Boolean) Add headers column to output virtual table
+- `add_key` (Boolean) Add key column to output virtual table
+- `add_timestamp` (Boolean) Add timestamp column to output virtual table
+- `keys_as_bytes` (Boolean) Make keys column as `bytes`, for non-utf8 characters
+- `value_as_bytes` (Boolean) Make value column as `bytes`, for non-utf8 characters
+
+
 <a id="nestedblock--settings--kinesis_source--parser--schema_registry"></a>
 ### Nested Schema for `settings.kinesis_source.parser.schema_registry`
 

internal/provider/transfer_endpoint_kafka.go

@@ -139,6 +139,7 @@ type endpointParser struct {
 	TSKV           *transferParserGeneric `tfsdk:"tskv"`
 	Blank          *blankParser           `tfsdk:"blank"`
 	SchemaRegistry *schemaRegistryParser  `tfsdk:"schema_registry"`
+	RawTable       *rawTableParser        `tfsdk:"raw_table"`
 }
 
 func endpointKafkaParserSchema() schema.Block {
@@ -148,44 +149,40 @@ func endpointKafkaParserSchema() schema.Block {
 			"tskv":            transferParserGenericSchema(),
 			"blank":           blankParserSchema(),
 			"schema_registry": schemaRegistrySchema(),
+			"raw_table":       rawTableParserSchema(),
 		},
 	}
 }
 
 func (m *endpointParser) parse(e *endpoint.Parser) diag.Diagnostics {
 	var diags diag.Diagnostics
 
-	switch p := e.GetParser().(type) {
-	case *endpoint.Parser_JsonParser:
+	resetParsers := func() {
+		m.JSON = nil
 		m.TSKV = nil
 		m.Blank = nil
+		m.RawTable = nil
 		m.SchemaRegistry = nil
-		if m.JSON == nil {
-			m.JSON = new(transferParserGeneric)
-		}
+	}
+	switch p := e.GetParser().(type) {
+	case *endpoint.Parser_RawTable:
+		resetParsers()
+		m.RawTable = new(rawTableParser)
+		diags.Append(m.RawTable.parse(p.RawTable)...)
+	case *endpoint.Parser_JsonParser:
+		resetParsers()
+		m.JSON = new(transferParserGeneric)
 		diags.Append(m.JSON.parse(p.JsonParser)...)
 	case *endpoint.Parser_TskvParser:
-		m.JSON = nil
-		m.Blank = nil
-		m.SchemaRegistry = nil
-		if m.TSKV == nil {
-			m.TSKV = new(transferParserGeneric)
-		}
+		resetParsers()
+		m.TSKV = new(transferParserGeneric)
 		diags.Append(m.TSKV.parse(p.TskvParser)...)
 	case *endpoint.Parser_BlankParser:
-		m.JSON = nil
-		m.TSKV = nil
-		m.SchemaRegistry = nil
-		if m.Blank == nil {
-			m.Blank = new(blankParser)
-		}
+		resetParsers()
+		m.Blank = new(blankParser)
 	case *endpoint.Parser_ConfluentSchemaRegistryParser:
-		m.JSON = nil
-		m.TSKV = nil
-		m.Blank = nil
-		if m.SchemaRegistry == nil {
-			m.SchemaRegistry = new(schemaRegistryParser)
-		}
+		resetParsers()
+		m.SchemaRegistry = new(schemaRegistryParser)
 		diags.Append(m.SchemaRegistry.parse(p.ConfluentSchemaRegistryParser)...)
 	default:
 		diags.Append(diag.NewErrorDiagnostic("unknown parser type", fmt.Sprintf("%v", e.GetParser())))
@@ -198,6 +195,10 @@ func (m *endpointParser) convert(r *endpoint.Parser) diag.Diagnostics {
 	var diags diag.Diagnostics
 
 	switch {
+	case m.RawTable != nil:
+		prsr := new(endpoint.RawTable)
+		diags.Append(m.RawTable.convert(prsr)...)
+		r.Parser = &endpoint.Parser_RawTable{RawTable: prsr}
 	case m.SchemaRegistry != nil:
 		prsr := new(endpoint.ConfluentSchemaRegistryParser)
 		diags.Append(m.SchemaRegistry.convert(prsr)...)
@@ -327,6 +328,61 @@ type transferParserGeneric struct {
 	AddRestColumn   types.Bool            `tfsdk:"add_rest_column"`
 }
 
+type rawTableParser struct {
+	ValueAsBytes types.Bool `tfsdk:"value_as_bytes"`
+	KeysAsBytes  types.Bool `tfsdk:"keys_as_bytes"`
+	AddTimestamp types.Bool `tfsdk:"add_timestamp"`
+	AddHeaders   types.Bool `tfsdk:"add_headers"`
+	AddKey       types.Bool `tfsdk:"add_key"`
+}
+
+func (p *rawTableParser) parse(table *endpoint.RawTable) diag.Diagnostics {
+	var diags diag.Diagnostics
+	p.ValueAsBytes = types.BoolValue(table.ValueAsBytes)
+	p.KeysAsBytes = types.BoolValue(table.KeysAsBytes)
+	p.AddTimestamp = types.BoolValue(table.AddTimestamp)
+	p.AddHeaders = types.BoolValue(table.AddHeaders)
+	p.AddKey = types.BoolValue(table.AddKey)
+	return diags
+}
+
+func (p *rawTableParser) convert(r *endpoint.RawTable) diag.Diagnostics {
+	var diags diag.Diagnostics
+	r.KeysAsBytes = p.KeysAsBytes.ValueBool()
+	r.ValueAsBytes = p.ValueAsBytes.ValueBool()
+	r.AddTimestamp = p.AddTimestamp.ValueBool()
+	r.AddHeaders = p.AddHeaders.ValueBool()
+	r.AddKey = p.AddKey.ValueBool()
+	return diags
+}
+
+func rawTableParserSchema() schema.Block {
+	return schema.SingleNestedBlock{
+		Attributes: map[string]schema.Attribute{
+			"value_as_bytes": schema.BoolAttribute{
+				Optional:            true,
+				MarkdownDescription: "Make value column as `bytes`, for non-utf8 characters",
+			},
+			"keys_as_bytes": schema.BoolAttribute{
+				Optional:            true,
+				MarkdownDescription: "Make keys column as `bytes`, for non-utf8 characters",
+			},
+			"add_timestamp": schema.BoolAttribute{
+				Optional:            true,
+				MarkdownDescription: "Add timestamp column to output virtual table",
+			},
+			"add_headers": schema.BoolAttribute{
+				Optional:            true,
+				MarkdownDescription: "Add headers column to output virtual table",
+			},
+			"add_key": schema.BoolAttribute{
+				Optional:            true,
+				MarkdownDescription: "Add key column to output virtual table",
+			},
+		},
+	}
+}
+
 func blankParserSchema() schema.Block {
 	return schema.SingleNestedBlock{
 		Attributes: map[string]schema.Attribute{},