1 files changed, 70 insertions, 2 deletions
diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go
index 44e93d297..bc5121c87 100644
--- a/weed/s3api/s3api_object_handlers.go
+++ b/weed/s3api/s3api_object_handlers.go
@@ -4,6 +4,8 @@ import (
 	"crypto/md5"
 	"encoding/json"
 	"fmt"
+	"github.com/minio/minio/cmd"
+	"github.com/minio/minio/pkg/s3select"
 	"io"
 	"io/ioutil"
 	"net/http"
@@ -12,6 +14,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/server"
 	"github.com/gorilla/mux"
+	xhttp "github.com/minio/minio/cmd/http"
 )
 
 var (
@@ -115,10 +118,15 @@ func (s3a *S3ApiServer) DeleteMultipleObjectsHandler(w http.ResponseWriter, r *h
 }
 
 func (s3a *S3ApiServer) proxyToFiler(w http.ResponseWriter, r *http.Request, destUrl string, responseFn func(proxyResonse *http.Response, w http.ResponseWriter)) {
+	var method string
+	method = r.Method
 
-	glog.V(2).Infof("s3 proxying %s to %s", r.Method, destUrl)
+	if r.URL.RawQuery == "select&select-type=2" {
+		method = "GET"
+	}
+	glog.V(2).Infof("s3 proxying %s to %s", method, destUrl)
 
-	proxyReq, err := http.NewRequest(r.Method, destUrl, r.Body)
+	proxyReq, err := http.NewRequest(method, destUrl, r.Body)
 
 	if err != nil {
 		glog.Errorf("NewRequest %s: %v", destUrl, err)
@@ -224,3 +232,63 @@ func getObject(vars map[string]string) string {
 	}
 	return object
 }
+
+func (s3a *S3ApiServer) SelectObjectContent(w http.ResponseWriter, r *http.Request) {
+	vars := mux.Vars(r)
+	bucket := vars["bucket"]
+	object := getObject(vars)
+
+	s3Select, err := s3select.NewS3Select(r.Body)
+	if err != nil {
+		if serr, ok := err.(s3select.SelectError); ok {
+			encodedErrorResponse := encodeResponse(cmd.APIErrorResponse{
+				Code:       serr.ErrorCode(),
+				Message:    serr.ErrorMessage(),
+				BucketName: bucket,
+				Key:        object,
+				Resource:   r.URL.Path,
+				RequestID:  w.Header().Get(xhttp.AmzRequestID),
+				HostID:     "",
+			})
+			writeResponse(w, serr.HTTPStatusCode(), encodedErrorResponse, "application/xml")
+		} else {
+			writeErrorResponse(w, ErrInternalError, r.URL)
+		}
+		return
+	}
+
+	if strings.HasSuffix(r.URL.Path, "/") {
+		writeErrorResponse(w, ErrNotImplemented, r.URL)
+		return
+	}
+
+	destUrl := fmt.Sprintf("http://%s%s/%s%s",
+		s3a.option.Filer, s3a.option.BucketsPath, bucket, object)
+	passThroughResponseSelectObjectContent := func(proxyResonse *http.Response,
+		w http.ResponseWriter) {
+		getObject := func(offset, length int64) (io.ReadCloser, error) {
+			return proxyResonse.Body, nil
+		}
+		if err = s3Select.Open(getObject); err != nil {
+			if serr, ok := err.(s3select.SelectError); ok {
+				encodedErrorResponse := encodeResponse(cmd.APIErrorResponse{
+					Code:       serr.ErrorCode(),
+					Message:    serr.ErrorMessage(),
+					BucketName: bucket,
+					Key:        object,
+					Resource:   r.URL.Path,
+					RequestID:  w.Header().Get(xhttp.AmzRequestID),
+					HostID:     "",
+				})
+				writeResponse(w, serr.HTTPStatusCode(), encodedErrorResponse, mimeXML)
+			} else {
+				writeResponse(w, http.StatusInternalServerError, encodeResponse("not s3select.SelectError"), mimeXML)
+			}
+			return
+		}
+
+		s3Select.Evaluate(w)
+		s3Select.Close()
+	}
+	s3a.proxyToFiler(w, r, destUrl, passThroughResponseSelectObjectContent)
+}