Enable setting impdp specific options on import

This enables a selected subset of impdb specific options on import that
we know are safe to support more common import usages.

bug: 254697198
Change-Id: I5e0b8e8914c996f53319b822463e1f9a29e0126e

oracle/api/v1alpha1/import_types.go

@@ -45,6 +45,12 @@ type ImportSpec struct {
 	// Oracle Operator.
 	// +optional
 	GcsLogPath string `json:"gcsLogPath,omitempty"`
+
+	// Options is a map of options and their values for usage with the
+	// specified Import Type. Right now this is only supported for passing
+	// additional impdp specific options.
+	// +optional
+	Options map[string]string `json:"options,omitempty"`
 }
 
 // ImportStatus defines the observed state of Import.

oracle/config/crd/bases/oracle.db.anthosapis.com_imports.yaml

@@ -76,6 +76,13 @@ spec:
                 description: Instance is the resource name within same namespace to
                   import into.
                 type: string
+              options:
+                additionalProperties:
+                  type: string
+                description: Options is a map of options and their values for usage
+                  with the specified Import Type. Right now this is only supported
+                  for passing additional impdp specific options.
+                type: object
               type:
                 description: Type of the Import. If not specified, the default of
                   DataPump is assumed, which is the only supported option currently.

oracle/controllers/config_agent_helpers.go

@@ -1154,7 +1154,19 @@ type DataPumpImportRequest struct {
 	GcsPath string
 	// GCS path to output log file
 	GcsLogPath string
-	LroInput   *LROInput
+	// Additional command options from the user.
+	Options  map[string]string
+	LroInput *LROInput
+}
+
+var AllowedImpdpParams = map[string]bool{
+	"TABLE_EXISTS_ACTION": true,
+	"REMAP_TABLE":         true,
+	"REMAP_SCHEMA":        true,
+	"REMAP_TABLESPACE":    true,
+	"REMAP_DATAFILE":      true,
+	"PARALLEL":            true,
+	"NETWORK_LINK":        true,
 }
 
 // DataPumpImport imports data dump file provided in GCS path.
@@ -1167,17 +1179,26 @@ func DataPumpImport(ctx context.Context, r client.Reader, dbClientFactory Databa
 	}
 	defer func() { _ = closeConn() }()
 
+	commandParams := []string{
+		"FULL=YES",
+		"METRICS=YES",
+		"LOGTIME=ALL",
+	}
+	for k, v := range req.Options {
+		k = strings.ToUpper(k)
+		if _, found := AllowedImpdpParams[k]; found {
+			param := k + "=" + v
+			commandParams = append(commandParams, param)
+		}
+	}
+
 	return dbClient.DataPumpImportAsync(ctx, &dbdpb.DataPumpImportAsyncRequest{
 		SyncRequest: &dbdpb.DataPumpImportRequest{
-			PdbName:    req.PdbName,
-			DbDomain:   req.DbDomain,
-			GcsPath:    req.GcsPath,
-			GcsLogPath: req.GcsLogPath,
-			CommandParams: []string{
-				"FULL=YES",
-				"METRICS=YES",
-				"LOGTIME=ALL",
-			},
+			PdbName:       req.PdbName,
+			DbDomain:      req.DbDomain,
+			GcsPath:       req.GcsPath,
+			GcsLogPath:    req.GcsLogPath,
+			CommandParams: commandParams,
 		},
 		LroInput: &dbdpb.LROInput{
 			OperationId: req.LroInput.OperationId,

oracle/controllers/importcontroller/import_controller.go

@@ -175,6 +175,7 @@ func (r *ImportReconciler) handleNotStartedImport(ctx context.Context, log logr.
 			DbDomain:   inst.Spec.DBDomain,
 			GcsPath:    imp.Spec.GcsPath,
 			GcsLogPath: imp.Spec.GcsLogPath,
+			Options:    imp.Spec.Options,
 			LroInput:   &controllers.LROInput{OperationId: lroOperationID(imp)},
 		}
 		resp, err := controllers.DataPumpImport(ctx, r, r.DatabaseClientFactory, inst.Namespace, inst.Name, *dataPumpReq)

oracle/controllers/inttest/datapumptest/datapump_test.go

@@ -160,7 +160,8 @@ drop user scott cascade;`
 			sql = `alter session set container=pdb1;
 grant unlimited tablespace to scott;
 alter session set current_schema=scott;
-drop table test_table;`
+drop table test_table;
+create bigfile tablespace scotty;`
 			testhelpers.K8sExecuteSqlOrFail(pod, k8sEnv.CPNamespace, sql)
 
 			By("Importing Tables")
@@ -173,6 +174,10 @@ drop table test_table;`
 					Instance:     instanceName,
 					DatabaseName: "pdb1",
 					GcsPath:      tableExport.Spec.GcsPath,
+					Options: map[string]string{
+						"REMAP_TABLE":      "test_table:retest_table",
+						"REMAP_TABLESPACE": "USER:SCOTTY",
+					},
 				},
 			}
 			testhelpers.K8sCreateWithRetry(k8sEnv.K8sClient, k8sEnv.Ctx, tableImport)
@@ -184,7 +189,7 @@ drop table test_table;`
 					objKey, createdImport, k8s.Ready, metav1.ConditionTrue, k8s.ImportComplete, 5*time.Minute, k8s.FindConditionOrFailed)
 			}
 
-			testhelpers.VerifySimpleData(k8sEnv)
+			testhelpers.VerifySimpleDataRemapped(k8sEnv)
 		})
 	}
 

oracle/controllers/testhelpers/envtest.go

@@ -1033,6 +1033,15 @@ commit;`
 	Expect(out).To(Equal(""))
 }
 
+// VerifySimpleDataRemapped checks that the test row in 'retest_table' exists.
+func VerifySimpleDataRemapped(k8sEnv K8sOperatorEnvironment) {
+	pod := "mydb-sts-0"
+	sql := `alter session set container=pdb1;
+alter session set current_schema=scott;
+select name from retest_table;`
+	Expect(K8sExecuteSqlOrFail(pod, k8sEnv.DPNamespace, sql)).To(Equal("Hello World"))
+}
+
 // VerifySimpleData checks that the test row in 'pdb1' exists.
 func VerifySimpleData(k8sEnv K8sOperatorEnvironment) {
 	pod := "mydb-sts-0"

oracle/operator.yaml

@@ -1628,6 +1628,13 @@ spec:
                 description: Instance is the resource name within same namespace to
                   import into.
                 type: string
+              options:
+                additionalProperties:
+                  type: string
+                description: Options is a map of options and their values for usage
+                  with the specified Import Type. Right now this is only supported
+                  for passing additional impdp specific options.
+                type: object
               type:
                 description: Type of the Import. If not specified, the default of
                   DataPump is assumed, which is the only supported option currently.