s3-compatibility API: transition to dpq (datapath query)

* update top target `/s3` handler and its downstream methods
* remove the following MPT query names from the dpq exceptions list:
  - s3.QparamMptUploadID
  - s3.QparamMptUploads
  - s3.QparamMptPartNo
* (native multi-part API uses the same exact constants)
* unescape s3.QparamMptUploadID (and therefore, apc.QparamMptUploadID)
* micro-optimize: remove redundant dpq.alloc/parse (that would co-exist with
  query parsing in certain cases)
----
* unescape apc.QparamETLTransformArgs (fix)
* add dpq parsing unit test

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

ais/dpq.go

@@ -95,12 +95,6 @@ type (
 var _except = map[string]bool{
 	apc.QparamDontHeadRemote: false,
 
-	// flows that utilize the following query parameters perform conventional r.URL.Query()
-	// TODO -- FIXME: multipart must definitely transition to dpq
-	s3.QparamMptUploadID: false,
-	s3.QparamMptUploads:  false,
-	s3.QparamMptPartNo:   false,
-
 	s3.QparamAccessKeyID: false,
 	s3.QparamExpires:     false,
 	s3.QparamSignature:   false,
@@ -142,6 +136,11 @@ func dpqFree(d *dpq) {
 
 func (dpq *dpq) get(qparam string) string { return dpq.m[qparam] }
 
+func (dpq *dpq) has(qparam string) bool {
+	_, ok := dpq.m[qparam]
+	return ok
+}
+
 func (dpq *dpq) parse(rawQuery string) error {
 	var (
 		query = rawQuery // r.URL.RawQuery
@@ -217,8 +216,10 @@ func (dpq *dpq) parse(rawQuery string) error {
 			dpq.sv.sig = value // (base64.RawURLEncoding)
 
 		// Parameters that don't need unescaping
-		case apc.QparamMptUploadID, apc.QparamMptPartNo, apc.QparamFltPresence, apc.QparamBinfoWithOrWithoutRemote,
-			apc.QparamAppendType, apc.QparamETLName, apc.QparamETLTransformArgs,
+		case apc.QparamMptUploads, apc.QparamMptPartNo,
+			apc.QparamFltPresence, apc.QparamBinfoWithOrWithoutRemote,
+			apc.QparamETLName,
+			apc.QparamAppendType,
 			apc.QparamNewCustom,
 			apc.QparamKeepRemote,
 			apc.QparamTID:
@@ -234,7 +235,9 @@ func (dpq *dpq) parse(rawQuery string) error {
 				continue
 			}
 
-			// Default: unescape and store in map
+			// Default: unescape and store in map - e.g.:
+			// - apc.QparamMptUploadID == s3.QparamMptUploadID (case in point: s3 uploadId)
+			// - apc.QparamETLTransformArgs (arbitrary JSON encoded via url.Values)
 			dpq.m[key], err = _unescape(value)
 		}
 

ais/dpq_internal_test.go

@@ -0,0 +1,32 @@
+// Package ais: internal unit tests
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION. All rights reserved.
+ */
+package ais
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/NVIDIA/aistore/api/apc"
+	"github.com/NVIDIA/aistore/tools/tassert"
+)
+
+func TestDpqETLTransformArgs(t *testing.T) {
+	for _, args := range []string{
+		`{"value":"a+b%20&c"}`,
+		`hello world`,
+		`a/b?c=d&e=f`,
+	} {
+		q := url.Values{}
+		q.Set(apc.QparamETLTransformArgs, args)
+
+		dpq := dpqAlloc()
+		err := dpq.parse(q.Encode())
+		tassert.CheckFatal(t, err)
+
+		actual := dpq.get(apc.QparamETLTransformArgs)
+		tassert.Errorf(t, actual == args, "expected %q, got %q", args, actual)
+		dpqFree(dpq)
+	}
+}

ais/emd_internal_test.go

@@ -1,6 +1,6 @@
 // Package ais: internal unit tests
 /*
- * Copyright (c) 2021-2025, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2021-2026, NVIDIA CORPORATION. All rights reserved.
  */
 package ais
 

ais/multiproxy_internal_test.go

@@ -1,6 +1,6 @@
 // Package ais: internal unit tests
 /*
- * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.
  */
 package ais
 

ais/s3/const.go

@@ -1,10 +1,11 @@
 // Package s3 provides Amazon S3 compatibility layer
 /*
- * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.
  */
 package s3
 
 import (
+	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
 )
@@ -23,10 +24,12 @@ const (
 	QparamStartAfter        = "start-after"        // Start listing after this object key
 	QparamDelimiter         = "delimiter"          // Delimiter for grouping object keys
 
-	// multipart
-	QparamMptUploads        = "uploads"    // Start multipart upload or list active uploads
-	QparamMptUploadID       = "uploadId"   // Complete, abort, or list parts of specific multipart upload
-	QparamMptPartNo         = "partNumber" // Part number for multipart upload
+	// AIS native multipart APIs use
+	// canonical S3 constants
+	QparamMptUploads  = apc.QparamMptUploads  // Start multipart upload or list active uploads
+	QparamMptUploadID = apc.QparamMptUploadID // Complete, abort, or list parts of specific multipart upload
+	QparamMptPartNo   = apc.QparamMptPartNo   // Part number for multipart upload
+
 	QparamMptMaxUploads     = "max-uploads"
 	QparamMptUploadIDMarker = "upload-id-marker"
 

ais/test/multiproxy_test.go

@@ -1,6 +1,6 @@
 // Package integration_test.
 /*
- * Copyright (c) 2018-2025, NVIDIA CORPORATION. All rights reserved.
+ * Copyright (c) 2018-2026, NVIDIA CORPORATION. All rights reserved.
  */
 package integration_test
 

ais/tgts3.go

@@ -40,28 +40,41 @@ func (t *target) s3Handler(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
 	case http.MethodHead:
 		t.headObjS3(w, r, apiItems)
+		return
+	case http.MethodGet, http.MethodPut, http.MethodDelete, http.MethodPost:
+		// dpq parsed below
+	default:
+		cmn.WriteErr405(w, r, http.MethodDelete, http.MethodGet, http.MethodHead, http.MethodPut, http.MethodPost)
+		return
+	}
+
+	dpq := dpqAlloc()
+	defer dpqFree(dpq)
+	if err := dpq.parse(r.URL.RawQuery); err != nil {
+		s3.WriteErr(w, r, s3.ErrInfo{Err: err})
+		return
+	}
+	dpq.isS3 = true
+
+	switch r.Method {
 	case http.MethodGet:
-		t.getObjS3(w, r, apiItems)
+		t.getObjS3(w, r, apiItems, dpq)
 	case http.MethodPut:
-		config := cmn.GCO.Get()
-		t.putCopyMpt(w, r, config, apiItems)
+		t.putCopyMpt(w, r, cmn.GCO.Get(), dpq, apiItems)
 	case http.MethodDelete:
-		q := r.URL.Query()
-		if q.Has(s3.QparamMptUploadID) {
-			t.abortMptS3(w, r, apiItems, q)
+		if dpq.has(s3.QparamMptUploadID) {
+			t.abortMptS3(w, r, dpq, apiItems)
 		} else {
 			t.delObjS3(w, r, apiItems)
 		}
 	case http.MethodPost:
-		t.postObjS3(w, r, apiItems)
-	default:
-		cmn.WriteErr405(w, r, http.MethodDelete, http.MethodGet, http.MethodHead, http.MethodPut, http.MethodPost)
+		t.postObjS3(w, r, apiItems, dpq)
 	}
 }
 
 // PUT /s3/<bucket-name>/<object-name>
 // [switch] mpt | put | copy
-func (t *target) putCopyMpt(w http.ResponseWriter, r *http.Request, config *cmn.Config, items []string) {
+func (t *target) putCopyMpt(w http.ResponseWriter, r *http.Request, config *cmn.Config, dpq *dpq, items []string) {
 	cs := fs.Cap()
 	if cs.IsOOS() {
 		s3.WriteErr(w, r, s3.ErrInfo{Err: cs.Err(), Status: http.StatusInsufficientStorage})
@@ -77,9 +90,8 @@ func (t *target) putCopyMpt(w http.ResponseWriter, r *http.Request, config *cmn.
 		s3.HandleAwsChunked(r)
 	}
 
-	q := r.URL.Query()
 	switch {
-	case q.Has(s3.QparamMptPartNo) && q.Has(s3.QparamMptUploadID):
+	case dpq.has(s3.QparamMptPartNo) && dpq.has(s3.QparamMptUploadID):
 		if r.Header.Get(cos.S3HdrObjSrc) != "" {
 			// TODO:
 			// copy another object (or its range) => part of the specified multipart upload.
@@ -89,16 +101,16 @@ func (t *target) putCopyMpt(w http.ResponseWriter, r *http.Request, config *cmn.
 			return
 		}
 		if cmn.Rom.V(5, cos.ModS3) {
-			nlog.Infoln("putPartMpt", bck.String(), items, q)
+			nlog.Infoln("putPartMpt", bck.String(), items, dpq.m)
 		}
-		t.putPartMptS3(w, r, items, q, bck)
+		t.putPartMptS3(w, r, dpq, bck, items)
 	case r.Header.Get(cos.S3HdrObjSrc) == "":
 		objName, errN := s3.JoinValidateOname(w, r, items)
 		if errN != nil {
 			return
 		}
 		lom := core.AllocLOM(objName)
-		t.putObjS3(w, r, bck, config, lom)
+		t.putObjS3(w, r, bck, config, lom, dpq)
 		core.FreeLOM(lom)
 	default:
 		t.copyObjS3(w, r, config, items)
@@ -201,7 +213,7 @@ func (t *target) copyObjS3(w http.ResponseWriter, r *http.Request, config *cmn.C
 	sgl.Free()
 }
 
-func (t *target) putObjS3(w http.ResponseWriter, r *http.Request, bck *meta.Bck, config *cmn.Config, lom *core.LOM) {
+func (t *target) putObjS3(w http.ResponseWriter, r *http.Request, bck *meta.Bck, config *cmn.Config, lom *core.LOM, dpq *dpq) {
 	if err := lom.InitBck(bck); err != nil {
 		if cmn.IsErrRemoteBckNotFound(err) {
 			t.BMDVersionFixup(r)
@@ -215,14 +227,6 @@ func (t *target) putObjS3(w http.ResponseWriter, r *http.Request, bck *meta.Bck,
 	started := time.Now()
 	lom.SetAtimeUnix(started.UnixNano())
 
-	// TODO: dual checksumming, e.g. lom.SetCustom(apc.AWS, ...)
-
-	dpq := dpqAlloc()
-	if err := dpq.parse(r.URL.RawQuery); err != nil {
-		s3.WriteErr(w, r, s3.ErrInfo{Err: err})
-		dpqFree(dpq)
-		return
-	}
 	poi := allocPOI()
 	{
 		poi.atime = started.UnixNano()
@@ -240,26 +244,22 @@ func (t *target) putObjS3(w http.ResponseWriter, r *http.Request, bck *meta.Bck,
 	} else {
 		s3.SetS3Headers(w.Header(), lom)
 	}
-	dpqFree(dpq)
 }
 
 // GET s3/<bucket-name[/<object-name>]
-func (t *target) getObjS3(w http.ResponseWriter, r *http.Request, items []string) {
+func (t *target) getObjS3(w http.ResponseWriter, r *http.Request, items []string, dpq *dpq) {
 	bucket := items[0]
 	bck, ecode, err := meta.InitByNameOnly(bucket, t.owner.bmd)
 	if err != nil {
 		s3.WriteErr(w, r, s3.ErrInfo{Err: err, Status: ecode})
 		return
 	}
 
-	// TODO -- FIXME: transition to dpq
-	q := r.URL.Query()
-
-	if len(items) == 1 && q.Has(s3.QparamMptUploads) {
+	if len(items) == 1 && dpq.has(s3.QparamMptUploads) {
 		if cmn.Rom.V(5, cos.ModS3) {
-			nlog.Infoln("listUploadsMpt", bck.String(), q)
+			nlog.Infoln("listUploadsMpt", bck.String(), dpq.m)
 		}
-		t.listUploadsMptS3(w, bck, q)
+		t.listUploadsMptS3(w, bck, dpq)
 		return
 	}
 	if len(items) < 2 {
@@ -271,32 +271,25 @@ func (t *target) getObjS3(w http.ResponseWriter, r *http.Request, items []string
 	if errN != nil {
 		return
 	}
-	if q.Has(s3.QparamMptPartNo) {
+	if dpq.has(s3.QparamMptPartNo) {
 		if cmn.Rom.V(5, cos.ModS3) {
-			nlog.Infoln("getMptPart", bck.String(), objName, q)
+			nlog.Infoln("getMptPart", bck.String(), objName, dpq.m)
 		}
 		lom := core.AllocLOM(objName)
-		t.getPartMptS3(w, r, bck, lom, q)
+		t.getPartMptS3(w, r, bck, lom, dpq)
 		core.FreeLOM(lom)
 		return
 	}
-	uploadID := q.Get(s3.QparamMptUploadID)
+	uploadID := dpq.get(s3.QparamMptUploadID)
 	if uploadID != "" {
 		if cmn.Rom.V(5, cos.ModS3) {
-			nlog.Infoln("listPartsMpt", bck.String(), objName, q)
+			nlog.Infoln("listPartsMpt", bck.String(), objName, dpq.m)
 		}
-		t.listPartsMptS3(w, r, bck, objName, q)
+		t.listPartsMptS3(w, r, bck, objName, dpq)
 		return
 	}
 
-	dpq := dpqAlloc()
-	if err := dpq.parse(r.URL.RawQuery); err != nil {
-		dpqFree(dpq)
-		s3.WriteErr(w, r, s3.ErrInfo{Err: err})
-		return
-	}
 	lom := core.AllocLOM(objName)
-	dpq.isS3 = true
 	lom, err = t.getObject(w, r, dpq, bck, lom)
 	core.FreeLOM(lom)
 
@@ -307,7 +300,6 @@ func (t *target) getObjS3(w http.ResponseWriter, r *http.Request, items []string
 		}
 		s3.WriteErr(w, r, ei)
 	}
-	dpqFree(dpq)
 }
 
 // HEAD /s3/<bucket-name>/<object-name> (TODO: s3.HdrMptCnt)
@@ -417,27 +409,29 @@ func (t *target) delObjS3(w http.ResponseWriter, r *http.Request, items []string
 }
 
 // POST /s3/<bucket-name>/<object-name>
-func (t *target) postObjS3(w http.ResponseWriter, r *http.Request, items []string) {
+func (t *target) postObjS3(w http.ResponseWriter, r *http.Request, items []string, dpq *dpq) {
 	bck, ecode, err := meta.InitByNameOnly(items[0], t.owner.bmd)
 	if err != nil {
 		s3.WriteErr(w, r, s3.ErrInfo{Err: err, Status: ecode})
 		return
 	}
-	q := r.URL.Query()
-	if q.Has(s3.QparamMptUploads) {
+
+	if dpq.has(s3.QparamMptUploads) {
 		if cmn.Rom.V(5, cos.ModS3) {
-			nlog.Infoln("startMpt", bck.String(), items, q)
+			nlog.Infoln("startMpt", bck.String(), items, dpq.m)
 		}
-		t.startMptS3(w, r, items, bck)
+		t.startMptS3(w, r, bck, items)
 		return
 	}
-	if q.Has(s3.QparamMptUploadID) {
+
+	if dpq.has(s3.QparamMptUploadID) {
 		if cmn.Rom.V(5, cos.ModS3) {
-			nlog.Infoln("completeMpt", bck.String(), items, q)
+			nlog.Infoln("completeMpt", bck.String(), items, dpq.m)
 		}
-		t.completeMptS3(w, r, items, q, bck)
+		t.completeMptS3(w, r, dpq, bck, items)
 		return
 	}
+
 	err = fmt.Errorf("set query parameter %q to start multipart upload or %q to complete the upload",
 		s3.QparamMptUploads, s3.QparamMptUploadID)
 	s3.WriteErr(w, r, s3.ErrInfo{Err: err})