Added - Support for structured streaming for Data Flow

Matt Lukas · jotruon · commit 845b51338c1b · 2022-01-21T12:04:58.000-08:00
diff --git a/examples/dataflow/main.tf b/examples/dataflow/main.tf
@@ -91,6 +91,7 @@ resource "oci_dataflow_application" "tf_application" {
   #description     = var.application_description
   #freeform_tags   = var.application_freeform_tags
   #logs_bucket_uri = var.application_logs_bucket_uri}"
+  type             = "BATCH"
   archive_uri = var.application_archive_uri
   #parameters {
   #Required
diff --git a/internal/integrationtest/dataflow_application_test.go b/internal/integrationtest/dataflow_application_test.go
@@ -64,6 +64,7 @@ var (
 		"metastore_id":         acctest.Representation{RepType: acctest.Optional, Create: `${var.metastore_id}`},
 		"parameters":           acctest.RepresentationGroup{RepType: acctest.Optional, Group: applicationParametersRepresentation},
 		"private_endpoint_id":  acctest.Representation{RepType: acctest.Optional, Create: `${oci_dataflow_private_endpoint.test_private_endpoint.id}`},
+		"type":                 acctest.Representation{RepType: acctest.Optional, Create: `BATCH`},
 		"warehouse_bucket_uri": acctest.Representation{RepType: acctest.Optional, Create: `${var.dataflow_warehouse_bucket_uri}`},
 	}
 	applicationParametersRepresentation = map[string]interface{}{
@@ -169,6 +170,7 @@ func TestDataflowApplicationResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(resourceName, "state"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_updated"),
+				resource.TestCheckResourceAttr(resourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttr(resourceName, "warehouse_bucket_uri", warehouseBucketUri),
 
 				func(s *terraform.State) (err error) {
@@ -215,6 +217,7 @@ func TestDataflowApplicationResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(resourceName, "state"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_updated"),
+				resource.TestCheckResourceAttr(resourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttrSet(resourceName, "warehouse_bucket_uri"),
 
 				func(s *terraform.State) (err error) {
@@ -259,6 +262,7 @@ func TestDataflowApplicationResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(resourceName, "state"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_updated"),
+				resource.TestCheckResourceAttr(resourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttr(resourceName, "warehouse_bucket_uri", warehouseBucketUri),
 				resource.TestCheckResourceAttr(resourceName, "class_name", classNameUpdated),
 
@@ -294,6 +298,7 @@ func TestDataflowApplicationResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(datasourceName, "applications.0.state"),
 				resource.TestCheckResourceAttrSet(datasourceName, "applications.0.time_created"),
 				resource.TestCheckResourceAttrSet(datasourceName, "applications.0.time_updated"),
+				resource.TestCheckResourceAttr(datasourceName, "applications.0.type", "BATCH"),
 			),
 		},
 		// verify singular datasource
@@ -330,6 +335,7 @@ func TestDataflowApplicationResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(singularDatasourceName, "state"),
 				resource.TestCheckResourceAttrSet(singularDatasourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(singularDatasourceName, "time_updated"),
+				resource.TestCheckResourceAttr(singularDatasourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttr(singularDatasourceName, "warehouse_bucket_uri", warehouseBucketUri),
 			),
 		},
diff --git a/internal/integrationtest/dataflow_invoke_run_test.go b/internal/integrationtest/dataflow_invoke_run_test.go
@@ -59,6 +59,8 @@ var (
 		"metastore_id":         acctest.Representation{RepType: acctest.Optional, Create: `${var.metastore_id}`},
 		"num_executors":        acctest.Representation{RepType: acctest.Optional, Create: `1`},
 		"parameters":           acctest.RepresentationGroup{RepType: acctest.Optional, Group: invokeRunParametersRepresentation},
+		"spark_version":        acctest.Representation{RepType: acctest.Optional, Create: `sparkVersion`},
+		"type":                 acctest.Representation{RepType: acctest.Optional, Create: `BATCH`},
 		"warehouse_bucket_uri": acctest.Representation{RepType: acctest.Optional, Create: `${var.dataflow_warehouse_bucket_uri}`},
 	}
 	invokeRunParametersRepresentation = map[string]interface{}{
@@ -153,6 +155,7 @@ func TestDataflowInvokeRunResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(resourceName, "state"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_updated"),
+				resource.TestCheckResourceAttr(resourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttr(resourceName, "warehouse_bucket_uri", warehouseBucketUri),
 				resource.TestCheckResourceAttr(resourceName, "metastore_id", metastoreId),
 
@@ -199,6 +202,7 @@ func TestDataflowInvokeRunResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(resourceName, "state"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_updated"),
+				resource.TestCheckResourceAttr(resourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttrSet(resourceName, "warehouse_bucket_uri"),
 
 				func(s *terraform.State) (err error) {
@@ -237,6 +241,7 @@ func TestDataflowInvokeRunResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(resourceName, "state"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(resourceName, "time_updated"),
+				resource.TestCheckResourceAttr(resourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttr(resourceName, "warehouse_bucket_uri", warehouseBucketUri),
 
 				func(s *terraform.State) (err error) {
@@ -275,6 +280,7 @@ func TestDataflowInvokeRunResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(datasourceName, "runs.0.time_created"),
 				resource.TestCheckResourceAttrSet(datasourceName, "runs.0.time_updated"),
 				resource.TestCheckResourceAttrSet(datasourceName, "runs.0.total_ocpu"),
+				resource.TestCheckResourceAttr(datasourceName, "runs.0.type", "BATCH"),
 			),
 		},
 		// verify singular datasource
@@ -316,6 +322,7 @@ func TestDataflowInvokeRunResource_basic(t *testing.T) {
 				resource.TestCheckResourceAttrSet(singularDatasourceName, "time_created"),
 				resource.TestCheckResourceAttrSet(singularDatasourceName, "time_updated"),
 				resource.TestCheckResourceAttrSet(singularDatasourceName, "total_ocpu"),
+				resource.TestCheckResourceAttr(singularDatasourceName, "type", "BATCH"),
 				resource.TestCheckResourceAttr(singularDatasourceName, "warehouse_bucket_uri", warehouseBucketUri),
 				resource.TestCheckResourceAttr(singularDatasourceName, "metastore_id", metastoreId),
 			),
diff --git a/internal/service/dataflow/dataflow_application_data_source.go b/internal/service/dataflow/dataflow_application_data_source.go
@@ -158,6 +158,8 @@ func (s *DataflowApplicationDataSourceCrud) SetData() error {
 		s.D.Set("time_updated", s.Res.TimeUpdated.String())
 	}
 
+	s.D.Set("type", s.Res.Type)
+
 	if s.Res.WarehouseBucketUri != nil {
 		s.D.Set("warehouse_bucket_uri", *s.Res.WarehouseBucketUri)
 	}
diff --git a/internal/service/dataflow/dataflow_application_resource.go b/internal/service/dataflow/dataflow_application_resource.go
@@ -144,6 +144,12 @@ func DataflowApplicationResource() *schema.Resource {
 				Optional: true,
 				Computed: true,
 			},
+			"type": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+				ForceNew: true,
+			},
 			"warehouse_bucket_uri": {
 				Type:     schema.TypeString,
 				Optional: true,
@@ -362,6 +368,10 @@ func (s *DataflowApplicationResourceCrud) Create() error {
 		request.SparkVersion = &tmp
 	}
 
+	if type_, ok := s.D.GetOkExists("type"); ok {
+		request.Type = oci_dataflow.ApplicationTypeEnum(type_.(string))
+	}
+
 	if warehouseBucketUri, ok := s.D.GetOkExists("warehouse_bucket_uri"); ok {
 		tmp := warehouseBucketUri.(string)
 		request.WarehouseBucketUri = &tmp
@@ -646,6 +656,8 @@ func (s *DataflowApplicationResourceCrud) SetData() error {
 		s.D.Set("time_updated", s.Res.TimeUpdated.String())
 	}
 
+	s.D.Set("type", s.Res.Type)
+
 	if s.Res.WarehouseBucketUri != nil {
 		s.D.Set("warehouse_bucket_uri", *s.Res.WarehouseBucketUri)
 	}
diff --git a/internal/service/dataflow/dataflow_applications_data_source.go b/internal/service/dataflow/dataflow_applications_data_source.go
@@ -167,6 +167,8 @@ func (s *DataflowApplicationsDataSourceCrud) SetData() error {
 			application["time_updated"] = r.TimeUpdated.String()
 		}
 
+		application["type"] = r.Type
+
 		resources = append(resources, application)
 	}
 
diff --git a/internal/service/dataflow/dataflow_invoke_run_data_source.go b/internal/service/dataflow/dataflow_invoke_run_data_source.go
@@ -195,6 +195,8 @@ func (s *DataflowInvokeRunDataSourceCrud) SetData() error {
 		s.D.Set("total_ocpu", *s.Res.TotalOCpu)
 	}
 
+	s.D.Set("type", s.Res.Type)
+
 	if s.Res.WarehouseBucketUri != nil {
 		s.D.Set("warehouse_bucket_uri", *s.Res.WarehouseBucketUri)
 	}
diff --git a/internal/service/dataflow/dataflow_invoke_run_resource.go b/internal/service/dataflow/dataflow_invoke_run_resource.go
@@ -149,6 +149,12 @@ func DataflowInvokeRunResource() *schema.Resource {
 				Computed: true,
 				ForceNew: true,
 			},
+			"type": {
+				Type:     schema.TypeString,
+				Optional: true,
+				Computed: true,
+				ForceNew: true,
+			},
 			"warehouse_bucket_uri": {
 				Type:     schema.TypeString,
 				Optional: true,
@@ -446,6 +452,10 @@ func (s *DataflowInvokeRunResourceCrud) Create() error {
 		request.SparkVersion = &tmp
 	}
 
+	if type_, ok := s.D.GetOkExists("type"); ok {
+		request.Type = oci_dataflow.ApplicationTypeEnum(type_.(string))
+	}
+
 	if warehouseBucketUri, ok := s.D.GetOkExists("warehouse_bucket_uri"); ok {
 		tmp := warehouseBucketUri.(string)
 		request.WarehouseBucketUri = &tmp
@@ -673,6 +683,8 @@ func (s *DataflowInvokeRunResourceCrud) SetData() error {
 		s.D.Set("total_ocpu", *s.Res.TotalOCpu)
 	}
 
+	s.D.Set("type", s.Res.Type)
+
 	if s.Res.WarehouseBucketUri != nil {
 		s.D.Set("warehouse_bucket_uri", *s.Res.WarehouseBucketUri)
 	}
diff --git a/internal/service/dataflow/dataflow_invoke_runs_data_source.go b/internal/service/dataflow/dataflow_invoke_runs_data_source.go
@@ -214,6 +214,8 @@ func (s *DataflowInvokeRunsDataSourceCrud) SetData() error {
 			invokeRun["total_ocpu"] = *r.TotalOCpu
 		}
 
+		invokeRun["type"] = r.Type
+
 		resources = append(resources, invokeRun)
 	}
 
diff --git a/website/docs/d/dataflow_application.html.markdown b/website/docs/d/dataflow_application.html.markdown
@@ -61,5 +61,6 @@ The following attributes are exported:
 * `state` - The current state of this application. 
 * `time_created` - The date and time a application was created, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `time_updated` - The date and time a application was updated, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
+* `type` - The Spark application processing type. 
 * `warehouse_bucket_uri` - An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
diff --git a/website/docs/d/dataflow_applications.html.markdown b/website/docs/d/dataflow_applications.html.markdown
@@ -77,5 +77,6 @@ The following attributes are exported:
 * `state` - The current state of this application. 
 * `time_created` - The date and time a application was created, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `time_updated` - The date and time a application was updated, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
+* `type` - The Spark application processing type. 
 * `warehouse_bucket_uri` - An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
diff --git a/website/docs/d/dataflow_invoke_run.html.markdown b/website/docs/d/dataflow_invoke_run.html.markdown
@@ -71,5 +71,6 @@ The following attributes are exported:
 * `time_created` - The date and time a application was created, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `time_updated` - The date and time a application was updated, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `total_ocpu` - The total number of oCPU requested by the run. 
+* `type` - The Spark application processing type. 
 * `warehouse_bucket_uri` - An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
diff --git a/website/docs/d/dataflow_invoke_runs.html.markdown b/website/docs/d/dataflow_invoke_runs.html.markdown
@@ -91,5 +91,6 @@ The following attributes are exported:
 * `time_created` - The date and time a application was created, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `time_updated` - The date and time a application was updated, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `total_ocpu` - The total number of oCPU requested by the run. 
+* `type` - The Spark application processing type. 
 * `warehouse_bucket_uri` - An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
diff --git a/website/docs/r/dataflow_application.html.markdown b/website/docs/r/dataflow_application.html.markdown
@@ -44,6 +44,7 @@ resource "oci_dataflow_application" "test_application" {
 		value = var.application_parameters_value
 	}
 	private_endpoint_id = oci_dataflow_private_endpoint.test_private_endpoint.id
+	type = var.application_type
 	warehouse_bucket_uri = var.application_warehouse_bucket_uri
 }
 ```
@@ -74,6 +75,7 @@ The following arguments are supported:
 	* `value` - (Required) (Updatable) The value of the parameter. It must be a string of 0 or more characters of any kind. Examples: "" (empty string), "10", "mydata.xml", "${x}" 
 * `private_endpoint_id` - (Optional) (Updatable) The OCID of a private endpoint. 
 * `spark_version` - (Required) (Updatable) The Spark version utilized to run the application. 
+* `type` - (Optional) The Spark application processing type. 
 * `warehouse_bucket_uri` - (Optional) (Updatable) An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
 
@@ -112,6 +114,7 @@ The following attributes are exported:
 * `state` - The current state of this application. 
 * `time_created` - The date and time a application was created, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `time_updated` - The date and time a application was updated, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
+* `type` - The Spark application processing type. 
 * `warehouse_bucket_uri` - An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
 ## Timeouts
diff --git a/website/docs/r/dataflow_invoke_run.html.markdown b/website/docs/r/dataflow_invoke_run.html.markdown
@@ -40,6 +40,7 @@ resource "oci_dataflow_invoke_run" "test_invoke_run" {
 		value = var.invoke_run_parameters_value
 	}
 	spark_version = var.invoke_run_spark_version
+	type = var.invoke_run_type
 	warehouse_bucket_uri = var.invoke_run_warehouse_bucket_uri
 }
 ```
@@ -67,6 +68,7 @@ The following arguments are supported:
 	* `name` - (Required) The name of the parameter.  It must be a string of one or more word characters (a-z, A-Z, 0-9, _). Examples: "iterations", "input_file" 
 	* `value` - (Required) The value of the parameter. It must be a string of 0 or more characters of any kind. Examples: "" (empty string), "10", "mydata.xml", "${x}" 
 * `spark_version` - (Optional) The Spark version utilized to run the application. This value may be set if applicationId is not since the Spark version will be taken from the associated application. 
+* `type` - (Optional) The Spark application processing type. 
 * `warehouse_bucket_uri` - (Optional) An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
 
@@ -115,6 +117,7 @@ The following attributes are exported:
 * `time_created` - The date and time a application was created, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `time_updated` - The date and time a application was updated, expressed in [RFC 3339](https://tools.ietf.org/html/rfc3339) timestamp format. Example: `2018-04-03T21:10:29.600Z` 
 * `total_ocpu` - The total number of oCPU requested by the run. 
+* `type` - The Spark application processing type. 
 * `warehouse_bucket_uri` - An Oracle Cloud Infrastructure URI of the bucket to be used as default warehouse directory for BATCH SQL runs. See https://docs.cloud.oracle.com/iaas/Content/API/SDKDocs/hdfsconnector.htm#uriformat. 
 
 ## Timeouts

Original file line number	Diff line number	Diff line change
`@@ -158,6 +158,8 @@ func (s *DataflowApplicationDataSourceCrud) SetData() error {`
`158`	`158`	`s.D.Set("time_updated", s.Res.TimeUpdated.String())`
`159`	`159`	`}`
`160`	`160`
	`161`	`+ s.D.Set("type", s.Res.Type)`
	`162`	`+`
`161`	`163`	`if s.Res.WarehouseBucketUri != nil {`
`162`	`164`	`s.D.Set("warehouse_bucket_uri", *s.Res.WarehouseBucketUri)`
`163`	`165`	`}`
Original file line number	Diff line number	Diff line change
`@@ -167,6 +167,8 @@ func (s *DataflowApplicationsDataSourceCrud) SetData() error {`
`167`	`167`	`application["time_updated"] = r.TimeUpdated.String()`
`168`	`168`	`}`
`169`	`169`
	`170`	`+ application["type"] = r.Type`
	`171`	`+`
`170`	`172`	`resources = append(resources, application)`
`171`	`173`	`}`
`172`	`174`
Original file line number	Diff line number	Diff line change
`@@ -195,6 +195,8 @@ func (s *DataflowInvokeRunDataSourceCrud) SetData() error {`
`195`	`195`	`s.D.Set("total_ocpu", *s.Res.TotalOCpu)`
`196`	`196`	`}`
`197`	`197`
	`198`	`+ s.D.Set("type", s.Res.Type)`
	`199`	`+`
`198`	`200`	`if s.Res.WarehouseBucketUri != nil {`
`199`	`201`	`s.D.Set("warehouse_bucket_uri", *s.Res.WarehouseBucketUri)`
`200`	`202`	`}`
Original file line number	Diff line number	Diff line change
`@@ -214,6 +214,8 @@ func (s *DataflowInvokeRunsDataSourceCrud) SetData() error {`
`214`	`214`	`invokeRun["total_ocpu"] = *r.TotalOCpu`
`215`	`215`	`}`
`216`	`216`
	`217`	`+ invokeRun["type"] = r.Type`
	`218`	`+`
`217`	`219`	`resources = append(resources, invokeRun)`
`218`	`220`	`}`
`219`	`221`