Fetching hive tables from all schemas instead of tables from only

current schema.

Author: Vatsal Mevada

cluster/src/main/scala/io/snappydata/gemxd/ClusterCallbacksImpl.scala

@@ -22,7 +22,6 @@ import java.{lang, util}
 
 import scala.collection.mutable.ArrayBuffer
 import scala.util.Try
-import scala.util.control.NonFatal
 
 import com.gemstone.gemfire.distributed.internal.membership.InternalDistributedMember
 import com.gemstone.gemfire.internal.cache.{ExternalTableMetaData, GemFireCacheImpl}
@@ -37,20 +36,16 @@ import com.pivotal.gemfirexd.internal.snappy.{CallbackFactoryProvider, ClusterCa
 import io.snappydata.cluster.ExecutorInitiator
 import io.snappydata.impl.LeadImpl
 import io.snappydata.recovery.RecoveryService
-import io.snappydata.sql.catalog.{CatalogObjectType, SnappyExternalCatalog}
+import io.snappydata.sql.catalog.CatalogObjectType
 import io.snappydata.util.ServiceUtils
 import io.snappydata.{ServiceManager, SnappyEmbeddedTableStatsProviderService}
 
 import org.apache.spark.Logging
 import org.apache.spark.scheduler.cluster.SnappyClusterManager
 import org.apache.spark.serializer.{KryoSerializerPool, StructTypeSerializer}
-import org.apache.spark.sql.SaveMode
-import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat, CatalogTable}
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.execution.columnar.ExternalStoreUtils
-import org.apache.spark.sql.execution.columnar.ExternalStoreUtils.CaseInsensitiveMutableHashMap
-import org.apache.spark.sql.execution.datasources.DataSource
-import org.apache.spark.sql.internal.SnappySessionCatalog
-import org.apache.spark.sql.sources.DataSourceRegister
+import org.apache.spark.sql.{SaveMode, SnappyContext}
 
 /**
  * Callbacks that are sent by GemXD to Snappy for cluster management
@@ -263,11 +258,9 @@ object ClusterCallbacksImpl extends ClusterCallbacks with Logging {
     }
   }
 
-  override def getHiveTablesMetadata(connectionId: Long, schema: String):
+  override def getHiveTablesMetadata():
   util.Collection[ExternalTableMetaData] = {
-    val session = SnappySessionPerConnection.getSnappySessionForConnection(connectionId)
-    val catalogTables = session.sessionState.catalog.asInstanceOf[SnappySessionCatalog]
-        .getHiveCatalogTables(schema)
+    val catalogTables = SnappyContext.getHiveCatalogTables()
     import scala.collection.JavaConverters._
     getTablesMetadata(catalogTables).asJava
   }

core/src/main/scala/org/apache/spark/sql/SnappyContext.scala

@@ -50,6 +50,7 @@ import org.apache.spark.memory.MemoryManagerCallback
 import org.apache.spark.rdd.RDD
 import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd, SparkListenerExecutorAdded}
 import org.apache.spark.sql.catalyst.analysis.NoSuchTableException
+import org.apache.spark.sql.catalyst.catalog.CatalogTable
 import org.apache.spark.sql.catalyst.expressions.SortDirection
 import org.apache.spark.sql.collection.{ToolsCallbackInit, Utils}
 import org.apache.spark.sql.execution.columnar.ExternalStoreUtils.CaseInsensitiveMutableHashMap
@@ -875,6 +876,12 @@ object SnappyContext extends Logging {
     }
   }
 
+  def getHiveCatalogTables(skipSchemas: Seq[String] = "SYS" :: Nil): Seq[CatalogTable] = {
+    val catalog = hiveSession.sessionState.catalog
+    catalog.listDatabases().filter(s => skipSchemas.isEmpty || !skipSchemas.contains(s)).
+        flatMap(schema => catalog.listTables(schema).map(table => catalog.getTableMetadata(table)))
+  }
+
   private[spark] def getBlockIdIfNull(
       executorId: String): Option[BlockAndExecutorId] =
     Option(storeToBlockMap.get(executorId))

core/src/main/scala/org/apache/spark/sql/internal/SnappySessionCatalog.scala

@@ -897,12 +897,6 @@ class SnappySessionCatalog(val externalCatalog: SnappyExternalCatalog,
     } else super.listTables(schema, pattern)
   }
 
-  def getHiveCatalogTables(schema: String): Seq[CatalogTable] =
-    if (snappySession.enableHiveSupport) {
-      hiveSessionCatalog.listTables(schema)
-          .map(ti => hiveSessionCatalog.getTableMetadata(ti))
-    } else Seq.empty[CatalogTable]
-
   override def refreshTable(name: TableIdentifier): Unit = {
     val table = addMissingGlobalTempSchema(name)
     if (isTemporaryTable(table)) {