feat: auto-inject current schema for bare table names in replications

Added `CurrentSchema()` method to the Connection interface with a base
implementation that returns the "schema" property. Oracle connections
override this to query `SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA')` when
no schema property is set.

In `Config.Prepare()`, when a source stream specifies a table name
without a schema qualifier, the current schema is now automatically
resolved from the source connection and injected. This allows users to
use bare table names in replications without explicitly qualifying the
schema, improving usability especially for Oracle sources.

Author: flarco

core/dbio/database/database.go

@@ -77,6 +77,7 @@ type Connection interface {
 	CreateTable(tableName string, cols iop.Columns, tableDDL string) (err error)
 	CreateTemporaryTable(tableName string, cols iop.Columns) (err error)
 	CurrentDatabase() (string, error)
+	CurrentSchema() (string, error)
 	Db() *sqlx.DB
 	DbX() *DbX
 	DropTable(...string) error
@@ -1458,6 +1459,11 @@ func (conn *BaseConn) CurrentDatabase() (dbName string, err error) {
 	return
 }
 
+// CurrentSchema returns the name of the current schema
+func (conn *BaseConn) CurrentSchema() (schemaName string, err error) {
+	return conn.GetProp("schema"), nil
+}
+
 // GetDatabases returns databases for given connection
 func (conn *BaseConn) GetDatabases() (iop.Dataset, error) {
 	// fields: [name]

core/dbio/database/database_oracle.go

@@ -447,6 +447,23 @@ retry:
 	return ds.Count, err
 }
 
+// CurrentSchema returns the name of the current schema
+func (conn *OracleConn) CurrentSchema() (schemaName string, err error) {
+	if schemaName = conn.GetProp("schema"); schemaName != "" {
+		return schemaName, nil
+	}
+
+	data, err := conn.Query("SELECT SYS_CONTEXT('USERENV', 'CURRENT_SCHEMA') FROM dual")
+	if err != nil {
+		return schemaName, g.Error(err)
+	}
+
+	schemaName = cast.ToString(data.Rows[0][0])
+	conn.SetProp("schema", schemaName)
+
+	return schemaName, nil
+}
+
 func (conn *OracleConn) getColumnsString(ds *iop.Datastream, tgtColumns ...iop.Columns) string {
 	// build lookup of target column length by name (from GetColumns/metadata)
 	tgtLength := map[string]int{}

core/sling/config.go

@@ -820,6 +820,17 @@ func (cfg *Config) Prepare() (err error) {
 			if cfg.ReplicationStream != nil {
 				cfg.ReplicationStream.SQL = cfg.Source.Stream
 			}
+		} else if sTable.Name != "" && sTable.Schema == "" {
+			if dbConn, err := cfg.SrcConn.AsDatabase(); err == nil {
+				if schema, _ := dbConn.CurrentSchema(); schema != "" {
+					// normalize the injected schema's casing for the dialect
+					if schemaTable, perr := database.ParseTableName(schema+"."+sTable.Name, cfg.SrcConn.Type); perr == nil && schemaTable.Name != "" {
+						schema = schemaTable.Schema
+					}
+					sTable.Schema = schema
+					cfg.Source.Stream = sTable.FullName()
+				}
+			}
 		}
 	}
 

tests/replications/r.114.oracle_no_schema_columns.yaml

@@ -0,0 +1,87 @@
+source: '{my_source}'
+target: '{my_target}'
+
+defaults:
+  mode: full-refresh
+
+hooks:
+  start:
+    - type: query
+      connection: '{source.name}'
+      query: |
+        BEGIN
+          EXECUTE IMMEDIATE 'DROP TABLE {env.schema}.CUSTOMERS_NOSCHEMA PURGE';
+        EXCEPTION
+          WHEN OTHERS THEN
+            IF SQLCODE != -942 THEN
+              RAISE;
+            END IF;
+        END;
+
+    - type: query
+      connection: '{source.name}'
+      query: |
+        CREATE TABLE {env.schema}.CUSTOMERS_NOSCHEMA (
+          id NUMBER,
+          name VARCHAR2(50)
+        )
+
+    - type: query
+      connection: '{source.name}'
+      query: |
+        INSERT INTO {env.schema}.CUSTOMERS_NOSCHEMA (id, name) VALUES (1, 'Alice');
+        INSERT INTO {env.schema}.CUSTOMERS_NOSCHEMA (id, name) VALUES (2, 'Bob');
+        INSERT INTO {env.schema}.CUSTOMERS_NOSCHEMA (id, name) VALUES (3, 'Carol');
+
+    - type: query
+      connection: '{target.name}'
+      query: drop table if exists public.customers_noschema
+
+  end:
+    - type: query
+      connection: '{target.name}'
+      query: select * from public.customers_noschema order by id
+      into: result
+
+    - type: log
+      message: |
+        store.result => { pretty_table(store.result) }
+
+    # ensure all rows replicated with the bare (no-schema) stream name
+    - type: check
+      check: length(store.result) == 3
+
+    - type: check
+      check: store.result[0].name == "Alice"
+
+    - type: check
+      check: store.result[2].name == "Carol"
+
+    - type: query
+      connection: '{source.name}'
+      query: |
+        BEGIN
+          EXECUTE IMMEDIATE 'DROP TABLE {env.schema}.CUSTOMERS_NOSCHEMA PURGE';
+        EXCEPTION
+          WHEN OTHERS THEN
+            IF SQLCODE != -942 THEN
+              RAISE;
+            END IF;
+        END;
+
+    - type: query
+      connection: '{target.name}'
+      query: drop table if exists public.customers_noschema
+
+# bare table name (no schema qualifier) — exercises the schema-injection path.
+# Sling must inject & case-normalize the connection's default schema so Oracle's
+# owner-qualified metadata lookup resolves the columns.
+# https://github.com/slingdata-io/sling-cli/issues/749
+streams:
+  CUSTOMERS_NOSCHEMA:
+    object: public.customers_noschema
+
+env:
+  schema: ${SCHEMA}
+  my_source: ${SOURCE}
+  my_target: ${TARGET}

tests/suite.cli.yaml

@@ -2394,4 +2394,14 @@
     - 'step (group_inner_warn) failed'
     - 'group substep 3 still ran'
     - 'step (deep_warn) failed'
-    - 'all on_failure:warn status checks passed'
+    - 'all on_failure:warn status checks passed'
+
+- id: 249
+  name: Oracle bare table name injects schema for column lookup (https://github.com/slingdata-io/sling-cli/issues/749)
+  run: sling run -r tests/replications/r.114.oracle_no_schema_columns.yaml
+  env:
+    SCHEMA: ORACLE
+    SOURCE: ORACLE
+    TARGET: POSTGRES
+  output_contains:
+    - 'execution succeeded'