GH-994: Fix DatabaseMetaData NPEs when SqlInfo is unavailable (#995)

## What's Changed

Multiple DatabaseMetaData methods had NPEs when the method
`ArrowDatabaseMetadata.getSqlInfoAndCacheIfCacheIsEmpty(final SqlInfo
sqlInfoCommand, final Class<T> desiredType)`
returned null.

Now the method never returns null. If the database server does not
provide the requested info, either a sensible default is returned or a
SQLException is thrown.

Closes #994.

Author: ennuite

flight/flight-sql-jdbc-core/src/main/java/org/apache/arrow/driver/jdbc/ArrowDatabaseMetadata.java

@@ -45,6 +45,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.List;
@@ -86,9 +87,12 @@
 import org.apache.arrow.vector.util.Text;
 import org.apache.calcite.avatica.AvaticaConnection;
 import org.apache.calcite.avatica.AvaticaDatabaseMetaData;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Arrow Flight JDBC's implementation of {@link DatabaseMetaData}. */
 public class ArrowDatabaseMetadata extends AvaticaDatabaseMetaData {
+  private static final Logger LOGGER = LoggerFactory.getLogger(ArrowDatabaseMetadata.class);
   private static final String JAVA_REGEX_SPECIALS = "[]()|^-+*?{}$\\.";
   private static final Charset CHARSET = StandardCharsets.UTF_8;
   private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
@@ -774,7 +778,34 @@ private <T> T getSqlInfoAndCacheIfCacheIsEmpty(
         }
       }
     }
-    return desiredType.cast(cachedSqlInfo.get(sqlInfoCommand));
+    T value = desiredType.cast(cachedSqlInfo.get(sqlInfoCommand));
+    if (value != null) {
+      return value;
+    }
+    LOGGER.debug(
+        "SqlInfo {} not provided by server, returning default for type {}",
+        sqlInfoCommand.name(),
+        desiredType.getSimpleName());
+
+    // Return sensible defaults when SqlInfo is unavailable
+    if (desiredType == Long.class) {
+      return desiredType.cast(0L);
+    } else if (desiredType == Integer.class) {
+      return desiredType.cast(0);
+    } else if (desiredType == Boolean.class) {
+      return desiredType.cast(false);
+    } else if (desiredType == String.class) {
+      return desiredType.cast("");
+    } else if (desiredType == Map.class) {
+      return desiredType.cast(Collections.emptyMap());
+    } else if (desiredType == List.class) {
+      return desiredType.cast(Collections.emptyList());
+    }
+
+    throw new SQLException(
+        String.format(
+            "The value of the SqlInfo %s is null and it could not be cast to %s.",
+            sqlInfoCommand.name(), desiredType.getName()));
   }
 
   private Optional<String> convertListSqlInfoToString(final List<?> sqlInfoList) {

flight/flight-sql-jdbc-core/src/test/java/org/apache/arrow/driver/jdbc/ArrowDatabaseMetadataTest.java

@@ -1543,11 +1543,83 @@ public void testEmptySqlInfo() throws Exception {
     try (final Connection testConnection =
         FLIGHT_SERVER_EMPTY_SQLINFO_TEST_RULE.getConnection(false)) {
       final DatabaseMetaData metaData = testConnection.getMetaData();
+
       assertThat(metaData.getSQLKeywords(), is(""));
       assertThat(metaData.getNumericFunctions(), is(""));
       assertThat(metaData.getStringFunctions(), is(""));
       assertThat(metaData.getSystemFunctions(), is(""));
       assertThat(metaData.getTimeDateFunctions(), is(""));
+
+      assertThat(metaData.getMaxBinaryLiteralLength(), is(0));
+      assertThat(metaData.getMaxCharLiteralLength(), is(0));
+      assertThat(metaData.getMaxColumnNameLength(), is(0));
+      assertThat(metaData.getMaxColumnsInGroupBy(), is(0));
+      assertThat(metaData.getMaxColumnsInIndex(), is(0));
+      assertThat(metaData.getMaxColumnsInOrderBy(), is(0));
+      assertThat(metaData.getMaxColumnsInSelect(), is(0));
+      assertThat(metaData.getMaxColumnsInTable(), is(0));
+      assertThat(metaData.getMaxConnections(), is(0));
+      assertThat(metaData.getMaxCursorNameLength(), is(0));
+      assertThat(metaData.getMaxIndexLength(), is(0));
+      assertThat(metaData.getMaxSchemaNameLength(), is(0));
+      assertThat(metaData.getMaxProcedureNameLength(), is(0));
+      assertThat(metaData.getMaxCatalogNameLength(), is(0));
+      assertThat(metaData.getMaxRowSize(), is(0));
+      assertThat(metaData.getMaxStatementLength(), is(0));
+      assertThat(metaData.getMaxStatements(), is(0));
+      assertThat(metaData.getMaxTableNameLength(), is(0));
+      assertThat(metaData.getMaxTablesInSelect(), is(0));
+      assertThat(metaData.getMaxUserNameLength(), is(0));
+
+      assertThat(metaData.supportsColumnAliasing(), is(false));
+      assertThat(metaData.nullPlusNonNullIsNull(), is(false));
+      assertThat(metaData.supportsTableCorrelationNames(), is(false));
+      assertThat(metaData.supportsDifferentTableCorrelationNames(), is(false));
+      assertThat(metaData.supportsExpressionsInOrderBy(), is(false));
+      assertThat(metaData.supportsOrderByUnrelated(), is(false));
+      assertThat(metaData.supportsLikeEscapeClause(), is(false));
+      assertThat(metaData.supportsNonNullableColumns(), is(false));
+      assertThat(metaData.supportsIntegrityEnhancementFacility(), is(false));
+      assertThat(metaData.isCatalogAtStart(), is(false));
+      assertThat(metaData.supportsSelectForUpdate(), is(false));
+      assertThat(metaData.supportsStoredProcedures(), is(false));
+      assertThat(metaData.supportsCorrelatedSubqueries(), is(false));
+      assertThat(metaData.doesMaxRowSizeIncludeBlobs(), is(false));
+      assertThat(metaData.supportsTransactions(), is(false));
+      assertThat(metaData.dataDefinitionCausesTransactionCommit(), is(false));
+      assertThat(metaData.dataDefinitionIgnoredInTransactions(), is(false));
+      assertThat(metaData.supportsBatchUpdates(), is(false));
+      assertThat(metaData.supportsSavepoints(), is(false));
+      assertThat(metaData.supportsNamedParameters(), is(false));
+      assertThat(metaData.locatorsUpdateCopy(), is(false));
+      assertThat(metaData.supportsStoredFunctionsUsingCallSyntax(), is(false));
+      assertThat(metaData.supportsGroupBy(), is(false));
+      assertThat(metaData.supportsGroupByUnrelated(), is(false));
+      assertThat(metaData.supportsMinimumSQLGrammar(), is(false));
+      assertThat(metaData.supportsCoreSQLGrammar(), is(false));
+      assertThat(metaData.supportsExtendedSQLGrammar(), is(false));
+      assertThat(metaData.supportsANSI92EntryLevelSQL(), is(false));
+      assertThat(metaData.supportsANSI92IntermediateSQL(), is(false));
+      assertThat(metaData.supportsANSI92FullSQL(), is(false));
+      assertThat(metaData.supportsOuterJoins(), is(false));
+      assertThat(metaData.supportsFullOuterJoins(), is(false));
+      assertThat(metaData.supportsLimitedOuterJoins(), is(false));
+      assertThat(metaData.supportsSchemasInProcedureCalls(), is(false));
+      assertThat(metaData.supportsSchemasInIndexDefinitions(), is(false));
+      assertThat(metaData.supportsSchemasInPrivilegeDefinitions(), is(false));
+      assertThat(metaData.supportsCatalogsInIndexDefinitions(), is(false));
+      assertThat(metaData.supportsCatalogsInPrivilegeDefinitions(), is(false));
+      assertThat(metaData.supportsPositionedDelete(), is(false));
+      assertThat(metaData.supportsPositionedUpdate(), is(false));
+      assertThat(metaData.supportsSubqueriesInComparisons(), is(false));
+      assertThat(metaData.supportsSubqueriesInExists(), is(false));
+      assertThat(metaData.supportsSubqueriesInIns(), is(false));
+      assertThat(metaData.supportsSubqueriesInQuantifieds(), is(false));
+      assertThat(metaData.supportsUnion(), is(false));
+      assertThat(metaData.supportsUnionAll(), is(false));
+      assertThat(metaData.supportsConvert(), is(false));
+
+      assertThat(metaData.getDefaultTransactionIsolation(), is(Connection.TRANSACTION_NONE));
     }
   }
 }