Add table type filtering (#189)

Author: samikshya-db

src/main/java/com/databricks/jdbc/client/impl/helper/MetadataResultConstants.java

@@ -5,6 +5,7 @@
 import java.util.List;
 
 public class MetadataResultConstants {
+  public static final String[] DEFAULT_TABLE_TYPES = {"TABLE", "VIEW", "SYSTEM TABLE"};
   private static final ResultColumn CATALOG_COLUMN =
       new ResultColumn("TABLE_CAT", "catalogName", Types.VARCHAR);
   private static final ResultColumn TYPE_CATALOG_COLUMN =

src/main/java/com/databricks/jdbc/client/impl/helper/MetadataResultSetBuilder.java

@@ -35,8 +35,13 @@ public static DatabricksResultSet getSchemasResult(ResultSet resultSet) throws S
     return buildResultSet(SCHEMA_COLUMNS, rows, METADATA_STATEMENT_ID);
   }
 
-  public static DatabricksResultSet getTablesResult(ResultSet resultSet) throws SQLException {
-    List<List<Object>> rows = getRows(resultSet, TABLE_COLUMNS);
+  public static DatabricksResultSet getTablesResult(ResultSet resultSet, String[] tableTypes)
+      throws SQLException {
+    List<String> allowedTableTypes = List.of(tableTypes);
+    List<List<Object>> rows =
+        getRows(resultSet, TABLE_COLUMNS).stream()
+            .filter(row -> allowedTableTypes.contains(row.get(3))) // Filtering based on table type
+            .collect(Collectors.toList());
     return buildResultSet(TABLE_COLUMNS, rows, GET_TABLES_STATEMENT_ID);
   }
 

src/main/java/com/databricks/jdbc/client/impl/sdk/DatabricksNewMetadataSdkClient.java

@@ -1,5 +1,6 @@
 package com.databricks.jdbc.client.impl.sdk;
 
+import static com.databricks.jdbc.client.impl.helper.MetadataResultConstants.DEFAULT_TABLE_TYPES;
 import static com.databricks.jdbc.client.impl.sdk.ResultConstants.TYPE_INFO_RESULT;
 
 import com.databricks.jdbc.client.DatabricksMetadataClient;
@@ -62,12 +63,16 @@ public DatabricksResultSet listTables(
       String tableNamePattern,
       String[] tableTypes)
       throws SQLException {
+    tableTypes =
+        Optional.ofNullable(tableTypes)
+            .filter(types -> types.length > 0)
+            .orElse(DEFAULT_TABLE_TYPES);
     CommandBuilder commandBuilder =
         new CommandBuilder(catalog, session)
             .setSchemaPattern(schemaNamePattern)
             .setTablePattern(tableNamePattern);
     String SQL = commandBuilder.getSQLString(CommandName.LIST_TABLES);
-    return MetadataResultSetBuilder.getTablesResult(getResultSet(SQL, session));
+    return MetadataResultSetBuilder.getTablesResult(getResultSet(SQL, session), tableTypes);
   }
 
   @Override

src/test/java/com/databricks/jdbc/client/impl/DatabricksNewMetadataSdkClientTest.java

@@ -202,8 +202,11 @@ void testListTables(
         .thenReturn(mockedResultSet);
     when(mockedResultSet.next()).thenReturn(true, false);
     for (ResultColumn resultColumn : TABLE_COLUMNS) {
-      when(mockedResultSet.getObject(resultColumn.getResultSetColumnName()))
-          .thenReturn(TEST_COLUMN);
+      if (resultColumn == TABLE_COLUMNS.get(3)) {
+        when(mockedResultSet.getObject(resultColumn.getResultSetColumnName())).thenReturn("TABLE");
+      } else
+        when(mockedResultSet.getObject(resultColumn.getResultSetColumnName()))
+            .thenReturn(TEST_COLUMN);
     }
     DatabricksResultSet actualResult =
         metadataClient.listTables(session, catalog, schema, table, null);