fix(oracle): stream LOB values on legacy servers
This commit is contained in:
t8y2
parent
21100ec22b
commit
cbf707c478
|
|
@ -31,13 +31,26 @@ const defaultMaxRows = 1000
|
|||
const oracleCharsetZHS32GB18030 = 854
|
||||
const legacyAgentSessionID = "__legacy__"
|
||||
const maxAgentSessions = 256
|
||||
const oracleLegacyLOBMaxMajorVersion = 11
|
||||
const oracleDatabaseVersionProbeTimeout = 3 * time.Second
|
||||
|
||||
const oracleDatabaseVersionSQL = `
|
||||
SELECT VERSION
|
||||
FROM PRODUCT_COMPONENT_VERSION
|
||||
WHERE PRODUCT LIKE 'Oracle Database%'
|
||||
AND ROWNUM = 1`
|
||||
|
||||
var (
|
||||
oraclePlSQLBlockStartRegexp = regexp.MustCompile(`(?is)^\s*(?:DECLARE|BEGIN|CREATE\s+(?:OR\s+REPLACE\s+)?(?:(?:EDITIONABLE|NONEDITIONABLE)\s+)?(?:FUNCTION|PROCEDURE|TRIGGER|PACKAGE(?:\s+BODY)?|TYPE(?:\s+BODY)?))\b`)
|
||||
oraclePlSQLBlockEndRegexp = regexp.MustCompile(`(?is)\bEND\s*;\s*$`)
|
||||
oracleNamedPlSQLBlockEndRegexp = regexp.MustCompile(`(?is)\bEND\s+([A-Z0-9_$#]+)\s*;\s*$`)
|
||||
oracleUnsupportedServerCharsetRegexp = regexp.MustCompile(`server use charset with id: ([0-9]+).*not supported by the driver`)
|
||||
oracleStringConverters = map[int]converters.IStringConverter{
|
||||
oracleVersionNumberRegexp = regexp.MustCompile(`(?:^|[^0-9])([0-9]+)\.[0-9]+`)
|
||||
oracleDatabaseVersionQueries = []string{
|
||||
oracleDatabaseVersionSQL,
|
||||
`SELECT BANNER FROM V$VERSION WHERE BANNER LIKE 'Oracle Database%' AND ROWNUM = 1`,
|
||||
}
|
||||
oracleStringConverters = map[int]converters.IStringConverter{
|
||||
oracleCharsetZHS32GB18030: oracleGB18030Converter{},
|
||||
}
|
||||
)
|
||||
|
|
@ -793,7 +806,7 @@ func (s *server) dispatch(method string, params map[string]json.RawMessage) (any
|
|||
|
||||
func (s *server) connect(params connectParams) error {
|
||||
_ = s.disconnect()
|
||||
db, err := openAndPingDB(params, 15*time.Second)
|
||||
db, effectiveParams, err := openSessionDB(params, 15*time.Second)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
|
@ -804,10 +817,109 @@ func (s *server) connect(params connectParams) error {
|
|||
return err
|
||||
}
|
||||
s.db = db
|
||||
s.params = params
|
||||
s.params = effectiveParams
|
||||
return nil
|
||||
}
|
||||
|
||||
func openSessionDB(params connectParams, timeout time.Duration) (*sql.DB, connectParams, error) {
|
||||
db, err := openAndPingDB(params, timeout)
|
||||
if err != nil {
|
||||
return nil, params, err
|
||||
}
|
||||
if hasOracleLOBFetchOption(params) {
|
||||
return db, params, nil
|
||||
}
|
||||
majorVersion, ok := oracleServerMajorVersion(db, timeout)
|
||||
if !shouldUseLegacyOracleLOBFetch(params, majorVersion, ok) {
|
||||
return db, params, nil
|
||||
}
|
||||
|
||||
effectiveParams := withOracleLOBFetchPost(params)
|
||||
if effectiveParams == params {
|
||||
return db, params, nil
|
||||
}
|
||||
if err := db.Close(); err != nil {
|
||||
return nil, params, err
|
||||
}
|
||||
db, err = openAndPingDB(effectiveParams, timeout)
|
||||
if err != nil {
|
||||
return nil, params, err
|
||||
}
|
||||
return db, effectiveParams, nil
|
||||
}
|
||||
|
||||
func oracleServerMajorVersion(db *sql.DB, timeout time.Duration) (int, bool) {
|
||||
if timeout > oracleDatabaseVersionProbeTimeout {
|
||||
timeout = oracleDatabaseVersionProbeTimeout
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), timeout)
|
||||
defer cancel()
|
||||
for _, query := range oracleDatabaseVersionQueries {
|
||||
var version string
|
||||
err := db.QueryRowContext(ctx, query).Scan(&version)
|
||||
if err == nil {
|
||||
if majorVersion, ok := parseOracleMajorVersion(version); ok {
|
||||
return majorVersion, true
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0, false
|
||||
}
|
||||
|
||||
func parseOracleMajorVersion(version string) (int, bool) {
|
||||
match := oracleVersionNumberRegexp.FindStringSubmatch(strings.TrimSpace(version))
|
||||
if len(match) != 2 {
|
||||
return 0, false
|
||||
}
|
||||
value, err := strconv.Atoi(match[1])
|
||||
return value, err == nil && value > 0
|
||||
}
|
||||
|
||||
func shouldUseLegacyOracleLOBFetch(params connectParams, majorVersion int, versionKnown bool) bool {
|
||||
return versionKnown && majorVersion <= oracleLegacyLOBMaxMajorVersion && !hasOracleLOBFetchOption(params)
|
||||
}
|
||||
|
||||
func hasOracleLOBFetchOption(params connectParams) bool {
|
||||
connectionString := strings.TrimSpace(params.ConnectionString)
|
||||
if strings.HasPrefix(strings.ToLower(connectionString), "oracle://") {
|
||||
parsed, err := url.Parse(connectionString)
|
||||
return err == nil && hasURLValueKey(parsed.Query(), "LOB FETCH")
|
||||
}
|
||||
values, err := url.ParseQuery(params.URLParams)
|
||||
return err == nil && hasURLValueKey(values, "LOB FETCH")
|
||||
}
|
||||
|
||||
func hasURLValueKey(values url.Values, target string) bool {
|
||||
for key := range values {
|
||||
if strings.EqualFold(strings.TrimSpace(key), target) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func withOracleLOBFetchPost(params connectParams) connectParams {
|
||||
connectionString := strings.TrimSpace(params.ConnectionString)
|
||||
if strings.HasPrefix(strings.ToLower(connectionString), "oracle://") {
|
||||
parsed, err := url.Parse(connectionString)
|
||||
if err != nil {
|
||||
return params
|
||||
}
|
||||
values := parsed.Query()
|
||||
values.Set("LOB FETCH", "POST")
|
||||
parsed.RawQuery = values.Encode()
|
||||
params.ConnectionString = parsed.String()
|
||||
return params
|
||||
}
|
||||
values, err := url.ParseQuery(params.URLParams)
|
||||
if err != nil {
|
||||
return params
|
||||
}
|
||||
values.Set("LOB FETCH", "POST")
|
||||
params.URLParams = values.Encode()
|
||||
return params
|
||||
}
|
||||
|
||||
func (s *server) disconnect() error {
|
||||
s.closeAllQuerySessions()
|
||||
if s.db == nil {
|
||||
|
|
|
|||
|
|
@ -13,6 +13,8 @@ import (
|
|||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/sijms/go-ora/v2/configurations"
|
||||
)
|
||||
|
||||
func TestHandshakeResponse(t *testing.T) {
|
||||
|
|
@ -647,6 +649,154 @@ func TestBuildDSNAddsSysDbaOption(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
func TestParseOracleMajorVersion(t *testing.T) {
|
||||
tests := []struct {
|
||||
version string
|
||||
major int
|
||||
ok bool
|
||||
}{
|
||||
{version: "10.2.0.4.0", major: 10, ok: true},
|
||||
{version: "11.2.0.4.0 Production", major: 11, ok: true},
|
||||
{version: "Oracle Database 10g Enterprise Edition Release 10.2.0.4.0 - 64bit Production", major: 10, ok: true},
|
||||
{version: "19.0.0.0.0", major: 19, ok: true},
|
||||
{version: "", ok: false},
|
||||
{version: "Oracle Database 10g", ok: false},
|
||||
}
|
||||
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.version, func(t *testing.T) {
|
||||
major, ok := parseOracleMajorVersion(tt.version)
|
||||
if major != tt.major || ok != tt.ok {
|
||||
t.Fatalf("parseOracleMajorVersion(%q) = (%d, %t), want (%d, %t)", tt.version, major, ok, tt.major, tt.ok)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestOracleServerMajorVersionUsesProductComponentVersion(t *testing.T) {
|
||||
db, scripted := openOracleViewSourceTestDB(t, []oracleViewSourceQueryStep{
|
||||
{
|
||||
queryContains: "PRODUCT_COMPONENT_VERSION",
|
||||
args: []driver.Value{},
|
||||
rows: [][]driver.Value{{"10.2.0.4.0"}},
|
||||
},
|
||||
})
|
||||
|
||||
major, ok := oracleServerMajorVersion(db, time.Second)
|
||||
if !ok || major != 10 {
|
||||
t.Fatalf("oracleServerMajorVersion() = (%d, %t), want (10, true)", major, ok)
|
||||
}
|
||||
if scripted.next != 1 {
|
||||
t.Fatalf("expected one version query, got %d", scripted.next)
|
||||
}
|
||||
}
|
||||
|
||||
func TestOracleServerMajorVersionFallsBackToVersionBanner(t *testing.T) {
|
||||
db, scripted := openOracleViewSourceTestDB(t, []oracleViewSourceQueryStep{
|
||||
{
|
||||
queryContains: "PRODUCT_COMPONENT_VERSION",
|
||||
args: []driver.Value{},
|
||||
err: errors.New("view unavailable"),
|
||||
},
|
||||
{
|
||||
queryContains: "V$VERSION",
|
||||
args: []driver.Value{},
|
||||
rows: [][]driver.Value{{
|
||||
"Oracle Database 11g Enterprise Edition Release 11.2.0.4.0 - 64bit Production",
|
||||
}},
|
||||
},
|
||||
})
|
||||
|
||||
major, ok := oracleServerMajorVersion(db, time.Second)
|
||||
if !ok || major != 11 {
|
||||
t.Fatalf("oracleServerMajorVersion() = (%d, %t), want (11, true)", major, ok)
|
||||
}
|
||||
if scripted.next != 2 {
|
||||
t.Fatalf("expected both version queries, got %d", scripted.next)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWithOracleLOBFetchPostUsesURLParamsForGeneratedDSN(t *testing.T) {
|
||||
params := withOracleLOBFetchPost(connectParams{
|
||||
Host: "db.example.com",
|
||||
Port: 1521,
|
||||
Database: "ORCL",
|
||||
Username: "scott",
|
||||
Password: "tiger",
|
||||
URLParams: "CHARSET=ZHS16GBK",
|
||||
})
|
||||
|
||||
values, err := url.ParseQuery(params.URLParams)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if values.Get("LOB FETCH") != "POST" || values.Get("CHARSET") != "ZHS16GBK" {
|
||||
t.Fatalf("legacy LOB mode should preserve URL parameters, got: %s", params.URLParams)
|
||||
}
|
||||
config, err := configurations.ParseConfig(buildDSN(params))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if config.Lob != configurations.STREAM {
|
||||
t.Fatalf("generated DSN should enable streamed LOB reads, got: %s", buildDSN(params))
|
||||
}
|
||||
}
|
||||
|
||||
func TestWithOracleLOBFetchPostUpdatesRawOracleURL(t *testing.T) {
|
||||
params := withOracleLOBFetchPost(connectParams{
|
||||
ConnectionString: "oracle://scott:tiger@db.example.com:1521/ORCL?CHARSET=ZHS16GBK",
|
||||
})
|
||||
parsed, err := url.Parse(params.ConnectionString)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if parsed.Query().Get("LOB FETCH") != "POST" || parsed.Query().Get("CHARSET") != "ZHS16GBK" {
|
||||
t.Fatalf("raw Oracle URL should preserve query parameters, got: %s", params.ConnectionString)
|
||||
}
|
||||
config, err := configurations.ParseConfig(params.ConnectionString)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if config.Lob != configurations.STREAM {
|
||||
t.Fatalf("raw Oracle URL should enable streamed LOB reads, got: %s", params.ConnectionString)
|
||||
}
|
||||
}
|
||||
|
||||
func TestHasOracleLOBFetchOptionHonorsExplicitModes(t *testing.T) {
|
||||
tests := []connectParams{
|
||||
{URLParams: "lob+fetch=inline"},
|
||||
{URLParams: "LOB%20FETCH=POST"},
|
||||
{ConnectionString: "oracle://scott:tiger@db.example.com:1521/ORCL?lob+fetch=stream"},
|
||||
}
|
||||
for _, params := range tests {
|
||||
if !hasOracleLOBFetchOption(params) {
|
||||
t.Fatalf("explicit LOB fetch mode should be detected: %+v", params)
|
||||
}
|
||||
}
|
||||
if hasOracleLOBFetchOption(connectParams{URLParams: "CHARSET=ZHS16GBK"}) {
|
||||
t.Fatal("unrelated URL parameters should not be treated as an explicit LOB fetch mode")
|
||||
}
|
||||
}
|
||||
|
||||
func TestShouldUseLegacyOracleLOBFetchOnlyForLegacyServers(t *testing.T) {
|
||||
params := connectParams{URLParams: "CHARSET=ZHS16GBK"}
|
||||
if !shouldUseLegacyOracleLOBFetch(params, 10, true) {
|
||||
t.Fatal("Oracle 10g should use streamed LOB reads")
|
||||
}
|
||||
if !shouldUseLegacyOracleLOBFetch(params, 11, true) {
|
||||
t.Fatal("Oracle 11g should use streamed LOB reads")
|
||||
}
|
||||
if shouldUseLegacyOracleLOBFetch(params, 12, true) || shouldUseLegacyOracleLOBFetch(params, 19, true) {
|
||||
t.Fatal("modern Oracle versions should retain the driver's default LOB mode")
|
||||
}
|
||||
if shouldUseLegacyOracleLOBFetch(params, 0, false) {
|
||||
t.Fatal("unknown Oracle versions should retain the driver's default LOB mode")
|
||||
}
|
||||
if shouldUseLegacyOracleLOBFetch(connectParams{URLParams: "LOB+FETCH=INLINE"}, 10, true) {
|
||||
t.Fatal("an explicit user LOB mode should not be overridden")
|
||||
}
|
||||
}
|
||||
|
||||
func TestOracleGB18030ConverterRoundTrip(t *testing.T) {
|
||||
converter := oracleGB18030Converter{}
|
||||
input := "DBX \u4e2d\u6587 \U00020000"
|
||||
|
|
|
|||
Loading…
Reference in New Issue