SNOW-86947: Implement all converter

Author: sfc-gh-stakeda

arrow_iterator.pyx

@@ -4,8 +4,11 @@
 
 # distutils: language = c++
 
+from logging import getLogger
 from cpython.ref cimport PyObject
 
+logger = getLogger(__name__)
+
 cdef extern from "cpp/ArrowIterator/CArrowChunkIterator.hpp" namespace "sf":
     cdef cppclass CArrowChunkIterator:
         CArrowChunkIterator(PyObject* context)
@@ -18,19 +21,26 @@ cdef extern from "cpp/ArrowIterator/CArrowChunkIterator.hpp" namespace "sf":
 
 
 cdef class PyArrowChunkIterator:
-    cdef CArrowChunkIterator * cIterator
+    cdef CArrowChunkIterator* cIterator
+    cdef PyObject* cret
 
     def __cinit__(PyArrowChunkIterator self, object arrow_stream_reader, object arrow_context):
         self.cIterator = new CArrowChunkIterator(<PyObject*>arrow_context)
         for rb in arrow_stream_reader:
-            self.cIterator.addRecordBatch(<PyObject *>rb)
+            self.cIterator.addRecordBatch(<PyObject*>rb)
         self.cIterator.reset()
 
     def __dealloc__(PyArrowChunkIterator self):
         del self.cIterator
 
     def __next__(PyArrowChunkIterator self):
-        ret = <object>self.cIterator.nextRow()
+        cret = self.cIterator.nextRow()
+        if not cret:
+            logger.error("Internal error from CArrowChunkIterator\n")
+            # it looks like this line can help us get into python and detect the global variable immediately
+            # however, this log will not show up for unclear reason
+        ret = <object>cret
+
         if ret is None:
             raise StopIteration
         else:

cpp/ArrowIterator/CArrowChunkIterator.cpp

@@ -12,7 +12,7 @@
 #include "DateConverter.hpp"
 #include "TimeStampConverter.hpp"
 #include "TimeConverter.hpp"
-#include "logging.hpp"
+#include <string>
 
 namespace sf
 {
@@ -50,6 +50,10 @@ PyObject* CArrowChunkIterator::nextRow()
   if (m_rowIndexInBatch < m_rowCountInBatch)
   {
     this->currentRowAsTuple();
+    if (py::checkPyError())
+    {
+      return nullptr;
+    }
     return m_latestReturnedRow.get();
   }
   else
@@ -60,11 +64,19 @@ PyObject* CArrowChunkIterator::nextRow()
       m_rowIndexInBatch = 0;
       m_rowCountInBatch = m_cRecordBatches[m_currentBatchIndex]->num_rows();
       this->initColumnConverters();
+      if (py::checkPyError())
+      {
+        return nullptr;
+      }
 
       logger.info("Current batch index: %d, rows in current batch: %d",
                   m_currentBatchIndex, m_rowCountInBatch);
 
       this->currentRowAsTuple();
+      if (py::checkPyError())
+      {
+        return nullptr;
+      }
       return m_latestReturnedRow.get();
     }
   }
@@ -189,15 +201,22 @@ void CArrowChunkIterator::initColumnConverters()
 
           default:
           {
-            /** TODO: how to throw an exception will be decided later */
-            logger.error("unknown arrow internal data type(%d) for FIXED data",
-                         dt->id());
-            break;
+            std::string errorInfo = Logger::formatString(
+                "[Snowflake Exception] unknown arrow internal data type(%d) "
+                "for FIXED data",
+                dt->id());
+            logger.error(errorInfo.c_str());
+            PyErr_SetString(PyExc_Exception, errorInfo.c_str());
+            return;
           }
         }
         break;
       }
 
+      case SnowflakeType::Type::ANY:
+      case SnowflakeType::Type::CHAR:
+      case SnowflakeType::Type::OBJECT:
+      case SnowflakeType::Type::VARIANT:
       case SnowflakeType::Type::TEXT:
       {
         m_currentBatchConverters.push_back(
@@ -258,10 +277,13 @@ void CArrowChunkIterator::initColumnConverters()
 
           default:
           {
-            /** TODO: how to throw an exception will be decided later */
-            logger.error("unknown arrow internal data type(%d) for TIME data",
-                         dt->id());
-            break;
+            std::string errorInfo = Logger::formatString(
+                "[Snowflake Exception] unknown arrow internal data type(%d) "
+                "for TIME data",
+                dt->id());
+            logger.error(errorInfo.c_str());
+            PyErr_SetString(PyExc_Exception, errorInfo.c_str());
+            return;
           }
         }
         break;
@@ -292,11 +314,13 @@ void CArrowChunkIterator::initColumnConverters()
 
           default:
           {
-            /** TODO: how to throw an exception will be decided later */
-            logger.error(
-                "unknown arrow internal data type(%d) for TIMESTAMP_NTZ data",
+            std::string errorInfo = Logger::formatString(
+                "[Snowflake Exception] unknown arrow internal data type(%d) "
+                "for TIMESTAMP_NTZ data",
                 dt->id());
-            break;
+            logger.error(errorInfo.c_str());
+            PyErr_SetString(PyExc_Exception, errorInfo.c_str());
+            return;
           }
         }
         break;
@@ -327,11 +351,13 @@ void CArrowChunkIterator::initColumnConverters()
 
           default:
           {
-            /** TODO: how to throw an exception will be decided later */
-            logger.error(
-                "unknown arrow internal data type(%d) for TIMESTAMP_LTZ data",
+            std::string errorInfo = Logger::formatString(
+                "[Snowflake Exception] unknown arrow internal data type(%d) "
+                "for TIMESTAMP_LTZ data",
                 dt->id());
-            break;
+            logger.error(errorInfo.c_str());
+            PyErr_SetString(PyExc_Exception, errorInfo.c_str());
+            return;
           }
         }
         break;
@@ -366,11 +392,13 @@ void CArrowChunkIterator::initColumnConverters()
 
           default:
           {
-            /** TODO: how to throw an exception will be decided later */
-            logger.error(
-                "unknown arrow internal data type(%d) for TIMESTAMP_TZ data",
+            std::string errorInfo = Logger::formatString(
+                "[Snowflake Exception] unknown arrow internal data type(%d) "
+                "for TIMESTAMP_TZ data",
                 dt->id());
-            break;
+            logger.error(errorInfo.c_str());
+            PyErr_SetString(PyExc_Exception, errorInfo.c_str());
+            return;
           }
         }
 
@@ -379,10 +407,12 @@ void CArrowChunkIterator::initColumnConverters()
 
       default:
       {
-        /** TODO: how to throw an exception will be decided later */
-        logger.error("unknown snowflake data type : %d",
-                     metaData->value(metaData->FindKey("logicalType")));
-        break;
+        std::string errorInfo = Logger::formatString(
+            "[Snowflake Exception] unknown snowflake data type : %d",
+            metaData->value(metaData->FindKey("logicalType")));
+        logger.error(errorInfo.c_str());
+        PyErr_SetString(PyExc_Exception, errorInfo.c_str());
+        return;
       }
     }
   }

cpp/ArrowIterator/DateConverter.cpp

@@ -8,31 +8,22 @@ namespace sf
 {
 Logger DateConverter::logger("snowflake.connector.DateConverter");
 
-py::UniqueRef& DateConverter::m_pyDatetimeDate()
+py::UniqueRef& DateConverter::initPyDatetimeDate()
 {
   static py::UniqueRef pyDatetimeDate;
   if (pyDatetimeDate.empty())
   {
-    py::PyUniqueLock lock;
     py::UniqueRef pyDatetimeModule;
-    arrow::Status status = py::importPythonModule("datetime", pyDatetimeModule);
-    if (!status.ok())
-    {
-      /** TODO : How to throw an exception will be decided later */
-      logger.error("import python module 'datetime' failed");
-    }
-    status = py::importFromModule(pyDatetimeModule, "date", pyDatetimeDate);
-    if (!status.ok())
-    {
-      /** TODO : How to throw an exception will be decided later */
-      logger.error("import python module 'datetime.date' failed");
-    }
+    py::importPythonModule("datetime", pyDatetimeModule);
+    py::importFromModule(pyDatetimeModule, "date", pyDatetimeDate);
+    Py_XINCREF(pyDatetimeDate.get());
   }
   return pyDatetimeDate;
 }
 
 DateConverter::DateConverter(std::shared_ptr<arrow::Array> array)
-: m_array(std::dynamic_pointer_cast<arrow::Date32Array>(array))
+: m_array(std::dynamic_pointer_cast<arrow::Date32Array>(array)),
+  m_pyDatetimeDate(initPyDatetimeDate())
 {
 }
 
@@ -41,8 +32,7 @@ PyObject* DateConverter::toPyObject(int64_t rowIndex) const
   if (m_array->IsValid(rowIndex))
   {
     int32_t deltaDays = m_array->Value(rowIndex);
-    py::PyUniqueLock lock;
-    return PyObject_CallMethod(m_pyDatetimeDate().get(), "fromordinal", "i",
+    return PyObject_CallMethod(m_pyDatetimeDate.get(), "fromordinal", "i",
                                epochDay + deltaDays);
   }
   else

cpp/ArrowIterator/DateConverter.hpp

@@ -19,14 +19,16 @@ class DateConverter : public IColumnConverter
   PyObject* toPyObject(int64_t rowIndex) const override;
 
 private:
+  static py::UniqueRef& initPyDatetimeDate();
+
   std::shared_ptr<arrow::Date32Array> m_array;
 
   /** from Python Ordinal to 1970-01-01 */
   static constexpr int epochDay = 719163;
 
-  static py::UniqueRef& m_pyDatetimeDate();
-
   static Logger logger;
+
+  py::UniqueRef& m_pyDatetimeDate;
 };
 
 }  // namespace sf

cpp/ArrowIterator/DecimalConverter.cpp

@@ -9,19 +9,23 @@
 namespace sf
 {
 
-py::UniqueRef& DecimalBaseConverter::m_decimalConstructor()
+DecimalBaseConverter::DecimalBaseConverter()
+: m_pyDecimalConstructor(initPyDecimalConstructor())
 {
-  static py::UniqueRef decimalConstructor;
-  if (decimalConstructor.empty())
+}
+
+py::UniqueRef& DecimalBaseConverter::initPyDecimalConstructor()
+{
+  static py::UniqueRef pyDecimalConstructor;
+  if (pyDecimalConstructor.empty())
   {
-    py::PyUniqueLock lock;
     py::UniqueRef decimalModule;
-    arrow::Status status = py::importPythonModule("decimal", decimalModule);
-
-    status = py::importFromModule(decimalModule, "Decimal", decimalConstructor);
+    py::importPythonModule("decimal", decimalModule);
+    py::importFromModule(decimalModule, "Decimal", pyDecimalConstructor);
+    Py_XINCREF(pyDecimalConstructor.get());
   }
 
-  return decimalConstructor;
+  return pyDecimalConstructor;
 }
 
 DecimalFromDecimalConverter::DecimalFromDecimalConverter(
@@ -43,8 +47,7 @@ PyObject* DecimalFromDecimalConverter::toPyObject(int64_t rowIndex) const
 
     /** the reason we use c_str() instead of std::string here is that we may
      * meet some encoding problem with std::string */
-    py::PyUniqueLock lock;
-    return PyObject_CallFunction(m_decimalConstructor().get(), "s#",
+    return PyObject_CallFunction(m_pyDecimalConstructor.get(), "s#",
                                  formatDecimalString.c_str(),
                                  formatDecimalString.size());
   }

cpp/ArrowIterator/DecimalConverter.hpp

@@ -13,11 +13,14 @@ namespace sf
 class DecimalBaseConverter : public IColumnConverter
 {
 public:
-  DecimalBaseConverter() = default;
+  DecimalBaseConverter();
   virtual ~DecimalBaseConverter() = default;
 
 protected:
-  static py::UniqueRef& m_decimalConstructor();
+  py::UniqueRef& m_pyDecimalConstructor;
+
+private:
+  static py::UniqueRef& initPyDecimalConstructor();
 };
 
 class DecimalFromDecimalConverter : public DecimalBaseConverter
@@ -66,7 +69,7 @@ PyObject* DecimalFromIntConverter<T>::toPyObject(int64_t rowIndex) const
     int64_t val = m_array->Value(rowIndex);
 
     py::UniqueRef decimal(
-        PyObject_CallFunction(m_decimalConstructor().get(), "L", val));
+        PyObject_CallFunction(m_pyDecimalConstructor.get(), "L", val));
     return PyObject_CallMethod(decimal.get(), "scaleb", "i", -m_scale);
   }
   else

cpp/ArrowIterator/Python/Common.cpp

@@ -5,6 +5,9 @@
 
 namespace sf
 {
-/** this file is needed in the future, when the main functionality is done
- * e.g. our own Status data structure */
+
+namespace py
+{
+  // this file will be deleted if it is not used in the future
+}
 }  // namespace sf

cpp/ArrowIterator/Python/Common.hpp

@@ -5,12 +5,17 @@
 #define PC_PYTHON_COMMON_HPP
 
 #include <Python.h>
+#include "Util/macros.hpp"
 
 namespace sf
 {
 
 namespace py
 {
+inline bool checkPyError()
+{
+  return UNLIKELY(PyErr_Occurred());
+}
 
 /**
  * A RAII class to wrap the PyObject*. The semantics are like std::unique_ptr.