Implement select object

Selects and filters out data from stored object.
Client for select feature

Author: sinhaashish

docs/API.md

@@ -40,6 +40,7 @@ s3Client = Minio('s3.amazonaws.com',
 |                                                       | [`fput_object`](#fput_object)                           |                                                   |                                                                     |
 |                                                       | [`fget_object`](#fget_object)                           |                                                   |                                                                     |
 |                                                       | [`get_partial_object`](#get_partial_object)             |                                                   |                                                                     |
+|                                                       | [`select_object_content`](#select_object_content)             |                                                   |                                                                     |
 
 ## 1. Constructor
 
@@ -668,6 +669,70 @@ except ResponseError as err:
     print(err)
 ```
 
+<a name="select_object_content"></a>
+### select_object_content(self, bucket_name, object_name, obj)
+Select object content filters the contents of object based on a simple structured query language (SQL).
+
+__Parameters__
+
+|Param   |Type   |Description   |
+|:---|:---|:---|
+|``bucket_name``   |_string_   |Name of the bucket.   |
+|``object_name``   |_string_   |Name of the object.   |
+|``obj`` | _Object_ | Query Options   |
+
+
+__Return Value__
+
+|Param   |Type   |Description   |
+|:---|:---|:---|
+|``object``   | _urllib3.response.HTTPResponse_   |Represents http streaming reader.   |
+
+__Example__
+
+
+```py
+client = Minio('s3.amazonaws.com',
+               access_key='YOUR-ACCESSKEY',
+               secret_key='YOUR-SECRETKEY')
+
+obj = SelectObjectOptions(
+    expression="select * from s3object",
+    input_serialization = InputSerialization(compression_type="NONE",
+                              csv=CSVInput(FileHeaderInfo="USE",
+                                           RecordDelimiter="\n",
+                                           FieldDelimiter=",",
+                                           QuoteCharacter='"',
+                                           QuoteEscapeCharacter='"',
+                                           Comments="#",
+                                           AllowQuotedRecordDelimiter="FALSE"
+                                           ),
+                                ),
+    output_serialization=OutputSerialization(
+                    csv = CSVOutput(
+                        QuoteFields="ASNEEDED",
+                        RecordDelimiter="\n",
+                        FieldDelimiter=",",
+                        QuoteCharacter='"',
+                        QuoteEscapeCharacter='"',
+                        )
+                    ),
+    request_progress=RequestProgress(
+        enabled="TRUE"
+        )
+    )
+
+
+try:
+    records, stats = client.select_object_content('sinha', 'test.csv', obj)
+    with open('my-record', 'wb') as record_data:
+        record_data.write(records)
+    with open('my-stat', 'wb') as stat_data:        
+        stat_data.write(stats)
+except ResponseError as err:
+    print(err)
+```
+
 <a name="fget_object"></a>
 ### fget_object(bucket_name, object_name, file_path, request_headers=None)
 Downloads and saves the object as a file in the local filesystem.

examples/select_object_content.py

@@ -0,0 +1,87 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage, (C) 2019 MinIO, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+
+from minio import Minio
+from minio.error import ResponseError
+from minio import SelectObjectReader
+from minio.select_object_reader import CRCValidationError
+from minio.select_object_options import (SelectObjectOptions, CSVInput, JSONInput, RequestProgress, ParquetInput, InputSerialization, OutputSerialization, CSVOutput, JsonOutput)
+
+client = Minio('s3.amazonaws.com',
+               access_key='YOUR-ACCESSKEY',
+               secret_key='YOUR-SECRETKEY')
+
+obj = SelectObjectOptions(
+    expression=" select * from s3object",
+    input_serialization = InputSerialization(compression_type="NONE",
+                              csv=CSVInput(FileHeaderInfo="USE",
+                                           RecordDelimiter="\n",
+                                           FieldDelimiter=",",
+                                           QuoteCharacter='"',
+                                           QuoteEscapeCharacter='"',
+                                           Comments="#",
+                                           AllowQuotedRecordDelimiter="FALSE"
+                                           ),
+                             ),     
+                             # If input is JSON
+                             # json = JSONInput(Type="DOCUMENT"
+                             #                 )
+                             #         ),
+
+    output_serialization=OutputSerialization(
+                        csv = CSVOutput(
+                            QuoteFields="ASNEEDED",
+                            RecordDelimiter="\n",
+                            FieldDelimiter=",",
+                            QuoteCharacter='"',
+                            QuoteEscapeCharacter='"',
+                            )
+                        ),
+
+    # If output is JSON
+    # output_serialization=OutputSerialization(
+    #                 json = JsonOutput(
+    #                     RecordDelimiter="\n"
+    #                     )
+    #                 ),
+    request_progress=RequestProgress(
+        enabled="TRUE"
+        )
+    )
+
+try:
+    response_stream = client.select_object_content('<YOUR_BUCKET>', '<YOUR_OBJECT>', obj)
+    data = SelectObjectReader(response_stream)
+    file = open('my-record', 'w')
+    while True :
+        records = data.read(500)
+        file.write(records)    
+        if records == "" :
+            file.close()
+            break
+
+    # Get the stats of message
+    file = open('my-stat', 'w')
+    stat = ""
+    for k, v in data.stats_message().items(): 
+        stat += k + " : " + v + "\n"
+    file.write(stat)
+    file.close()
+
+except CRCValidationError as err:
+    print(err)
+except ResponseError as err:
+    print(err)

minio/__init__.py

@@ -31,10 +31,11 @@
 __author__ = 'MinIO, Inc.'
 __version__ = '4.0.19'
 __license__ = 'Apache 2.0'
-__copyright__ = 'Copyright 2015, 2016, 2017 MinIO, Inc.'
+__copyright__ = 'Copyright 2015, 2016, 2017, 2018, 2019 MinIO, Inc.'
 
 from .api import Minio
 from .error import ResponseError
 from .post_policy import PostPolicy
 from .copy_conditions import CopyConditions
 from .definitions import Bucket, Object
+from .select_object_reader import SelectObjectReader

minio/api.py

@@ -76,7 +76,7 @@
                       is_valid_source_sse_object,
                       is_valid_bucket_notification_config, is_valid_policy_type,
                       mkdir_p, dump_http, amzprefix_user_metadata,
-                      is_supported_header,is_amz_header)
+                      is_supported_header,is_amz_header,is_valid_input_type)
 from .helpers import (MAX_MULTIPART_OBJECT_SIZE,
                       MAX_PART_SIZE,
                       MAX_POOL_SIZE,
@@ -90,7 +90,8 @@
 from .xml_marshal import (xml_marshal_bucket_constraint,
                           xml_marshal_complete_multipart_upload,
                           xml_marshal_bucket_notifications,
-                          xml_marshal_delete_objects)
+                          xml_marshal_delete_objects,
+                          xml_marshal_select)
 from .fold_case_dict import FoldCaseDict
 from .thread_pool import ThreadPool
 
@@ -235,6 +236,40 @@ def trace_off(self):
         """
         self._trace_output_stream = None
 
+
+    def select_object_content(self, bucket_name,
+                        object_name,
+                        obj):
+
+        '''
+        Select from object
+        '''        
+        is_valid_bucket_name(bucket_name)
+        is_non_empty_string(object_name)
+        is_valid_input_type(object_name)
+
+        content = xml_marshal_select(obj)
+        url_values = dict()
+        url_values["select"] = ""
+        url_values["select-type"] = "2"
+
+        headers = {
+            'Content-Length': str(len(content)),
+            'Content-Md5': get_md5_base64digest(content)
+        }
+        content_sha256_hex = get_sha256_hexdigest(content)
+        return  self._url_open(
+            'POST',
+            bucket_name=bucket_name,
+            object_name=object_name,
+            query=url_values,
+            headers=headers,
+            body=content,
+            content_sha256=content_sha256_hex,
+            preload_content=False)
+
+          
+
     # Bucket level
     def make_bucket(self, bucket_name, location='us-east-1'):
         """

minio/helpers.py

@@ -55,6 +55,15 @@
 MIN_PART_SIZE = 5 * 1024 * 1024  # 5MiB
 DEFAULT_PART_SIZE = MIN_PART_SIZE # Currently its 5MiB
 
+READ_SIZE_SELECT = 32 * 1024 # Buffer size for select object content
+SQL = 'SQL'
+RECORDS = 'Records'
+PROGRESS = 'Progress'
+STATS ='Stats'
+EVENT = 'event'
+END = 'End'
+ERROR = 'error'
+
 _VALID_BUCKETNAME_REGEX = re.compile('^[a-z0-9][a-z0-9\\.\\-]+[a-z0-9]$')
 _ALLOWED_HOSTNAME_REGEX = re.compile(
     '^((?!-)(?!_)[A-Z_\\d-]{1,63}(?<!-)(?<!_)\\.)*((?!_)(?!-)[A-Z_\\d-]{1,63}(?<!-)(?<!_))$',
@@ -696,3 +705,18 @@ def is_supported_header(key):
 # returns true if header is a storage class header
 def is_storageclass_header(key):
     return key.lower() == "x-amz-storage-class"
+
+ # return the input type if valid i.e. csv, json,  parquet, par 
+ # or raise an InvalidArgumentError 
+def is_valid_input_type(object_name):
+    if  object_name.__contains__('.csv')  |  object_name.__contains__('.CSV') :
+        return 'csv'
+    elif  object_name.__contains__('.json') | object_name.__contains__('.JSON') :
+        return 'json'
+    elif   object_name.__contains__('.parquet') | object_name.__contains__('.par') :
+        return 'parquet'
+    else:
+        raise InvalidArgumentError('Input type not supported '
+                                   ' Only csv, json and parquet.')    
+
+

minio/select_object_options.py

@@ -0,0 +1,112 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage, (C) 2019 MinIO, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""
+
+This module cretaes the request for Select
+
+:copyright: (c) 2019 by MinIO, Inc.
+:license: Apache 2.0, see LICENSE for more details.
+
+"""
+from .helpers import (SQL)
+
+class CSVInput:
+    """CSVInput.
+
+    """
+    def __init__(self, FileHeaderInfo=None, RecordDelimiter="\n", 
+                 FieldDelimiter=",", QuoteCharacter='"',
+                 QuoteEscapeCharacter='"', Comments="#",
+                 AllowQuotedRecordDelimiter=False):
+        self.FileHeaderInfo = FileHeaderInfo
+        self.RecordDelimiter = RecordDelimiter
+        self.FieldDelimiter = FieldDelimiter
+        self.QuoteCharacter = QuoteCharacter
+        self.QuoteEscapeCharacter = QuoteEscapeCharacter
+        self.Comments = Comments
+        self.AllowQuotedRecordDelimiter = AllowQuotedRecordDelimiter
+
+class JSONInput:
+    """JSONInput.
+
+    """
+    def __init__(self, Type=None):
+        self.Type = Type
+
+class ParquetInput:
+    """ParquetInput.
+
+    """
+class InputSerialization:
+    """InputSerialization.
+
+    """
+    def __init__(self, compression_type="NONE", csv=None, json=None, par=None ):
+        self.compression_type = compression_type
+        self.csv_input = csv
+        self.json_input = json
+        self.parquet_input = par
+
+
+
+class CSVOutput:
+    """CSVOutput.
+
+    """
+    def __init__(self, QuoteFields="ASNEEDED", RecordDelimiter="\n", 
+                 FieldDelimiter=",", QuoteCharacter='"',
+                 QuoteEscapeCharacter='"'):
+        self.QuoteFields = QuoteFields
+        self.RecordDelimiter = RecordDelimiter
+        self.FieldDelimiter = FieldDelimiter
+        self.QuoteCharacter = QuoteCharacter
+        self.QuoteEscapeCharacter = QuoteEscapeCharacter
+
+class JsonOutput:
+    """
+    JsonOutput.
+
+    """
+    def __init__(self, RecordDelimiter="\n"):
+        self.RecordDelimiter = RecordDelimiter
+
+
+
+class OutputSerialization:
+    """OutputSerialization.
+
+    """
+    def __init__(self,  csv=None, json=None ):
+        self.csv_output = csv
+        self.json_output = json
+
+class RequestProgress:
+    """RequestProgress.
+
+    """
+    def __init__(self, enabled=False):
+        self.enabled = enabled
+
+class  SelectObjectOptions:
+    """SelectObjectOptions.
+
+    """
+    expression_type = SQL
+    def __init__(self, expression, input_serialization, output_serialization, request_progress):
+        self.expression = expression
+        self.in_ser = input_serialization
+        self.out_ser = output_serialization
+        self.req_progress = request_progress