Support both old and new catalog (#94)

In Spark 3.1 or earlier, the info_name field was named database_description_item and the info_value field was named database_description_value for the builtin catalog.

sparkly/catalog.py

@@ -232,17 +232,27 @@ def get_database_properties(self, db_name):
         Returns:
             dict[str,str]: Key/value for properties.
         """
+        describe = self._spark.sql(f'DESCRIBE DATABASE EXTENDED {db_name}')
+
+        if 'database_description_item' in describe.columns:
+            key_col = 'database_description_item'
+            val_col = 'database_description_value'
+        else:
+            key_col = 'info_name'
+            val_col = 'info_value'
+
         properties = (
             self._spark.sql('DESCRIBE DATABASE EXTENDED {}'.format(db_name))
-            .where(F.col('info_name') == 'Properties')
-            .select('info_value')
+            .where(F.col(key_col) == 'Properties')
+            .select(val_col)
             .first()
         )
 
         parsed_properties = {}
 
         if properties:
-            for name, value in read_db_properties_format(properties.info_value):
+            info_value = getattr(properties, val_col)
+            for name, value in read_db_properties_format(info_value):
                 parsed_properties[name] = value
 
         return parsed_properties

tests/integration/base.py

@@ -50,3 +50,9 @@ class SparklyTestSession(SparklySession):
         # will be overwritten by additional_options passed in setup_session
         'my.custom.option.3': '319',
     }
+
+
+class SparklyTestSessionWithOldCatalog(SparklyTestSession):
+    options = {
+        'spark.sql.legacy.keepCommandOutputSchema': 'true',
+    }

tests/integration/test_catalog.py

@@ -15,7 +15,7 @@
 #
 
 from sparkly.testing import SparklyGlobalSessionTest
-from tests.integration.base import SparklyTestSession
+from tests.integration.base import SparklyTestSession, SparklyTestSessionWithOldCatalog
 from sparkly.catalog import read_db_properties_format
 
 
@@ -197,3 +197,7 @@ def test_read_db_properties_format_for_broken_input(self):
 
         with self.assertRaises(ValueError):
             read_db_properties_format(')')
+
+
+class TestSparklyWithOldCatalog(TestSparklyCatalog):
+    session = SparklyTestSessionWithOldCatalog