[WIP] added a catalog API which can be extended by any external catalog

core/src/main/java/org/apache/carbondata/core/catalog/BaseCatalog.java

@@ -0,0 +1,31 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.core.catalog;
+
+import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
+import org.apache.carbondata.format.TableInfo;
+
+public interface BaseCatalog {
+
+  long getLastSchemaModificationTime(String schemaFilePath);
+
+  long saveSchema(AbsoluteTableIdentifier identifier, TableInfo thriftTableInfo);
+
+  TableInfo getSchema(String schemaFilePath);
+
+}

core/src/main/java/org/apache/carbondata/core/catalog/CatalogFactory.java

@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.core.catalog;
+
+import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.constants.CarbonCommonConstants;
+import org.apache.carbondata.core.util.CarbonProperties;
+
+import org.apache.log4j.Logger;
+
+public class CatalogFactory {
+
+  private static final Logger LOGGER =
+      LogServiceFactory.getLogService(CatalogFactory.class.getName());
+
+  private static final CatalogFactory INSTANCE = new CatalogFactory();
+
+  private Object catalog;
+
+  private CatalogFactory() {
+    String catalogClass = CarbonProperties.getInstance()
+        .getProperty(CarbonCommonConstants.CARBON_CATALOG_IMPL,
+            CarbonCommonConstants.CARBON_CATALOG_IMPL_DEFAULT);
+    try {
+      catalog = Class.forName(catalogClass).newInstance();
+    } catch (Exception e) {
+      LOGGER.error("Error while loading class: " + catalogClass);
+    }
+  }
+
+  public static CatalogFactory getInstance() {
+    return INSTANCE;
+  }
+
+  public BaseCatalog getCatalog() {
+    return ((BaseCatalog) catalog);
+  }
+
+  public <T extends BaseCatalog> T getCatalog(Class<T> type) {
+    return type.cast(catalog);
+  }
+}

core/src/main/java/org/apache/carbondata/core/constants/CarbonCommonConstants.java

@@ -2876,4 +2876,10 @@ private CarbonCommonConstants() {
 
   public static final String CARBON_ENABLE_SCHEMA_ENFORCEMENT_DEFAULT = "true";
 
+  @CarbonProperty
+  public static final String CARBON_CATALOG_IMPL = "carbon.catalog.impl";
+
+  @CarbonProperty public static final String CARBON_CATALOG_IMPL_DEFAULT =
+      "org.apache.spark.sql.catalog.CarbonCatalogImpl";
+
 }

core/src/main/java/org/apache/carbondata/core/metadata/converter/SchemaConverter.java

@@ -61,8 +61,7 @@ org.apache.carbondata.format.TableSchema fromWrapperToExternalTableSchema(
    * @param tableName
    * @return
    */
-  org.apache.carbondata.format.TableInfo fromWrapperToExternalTableInfo(TableInfo wrapperTableInfo,
-      String dbName, String tableName);
+  org.apache.carbondata.format.TableInfo fromWrapperToExternalTableInfo(TableInfo wrapperTableInfo);
 
   /**
    * @param externalSchemaEvolutionEntry

core/src/main/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImpl.java

@@ -306,7 +306,7 @@ private org.apache.carbondata.format.BucketingInfo fromWrapperToExternalBucketin
    */
   @Override
   public org.apache.carbondata.format.TableInfo fromWrapperToExternalTableInfo(
-      TableInfo wrapperTableInfo, String dbName, String tableName) {
+      TableInfo wrapperTableInfo) {
     org.apache.carbondata.format.TableSchema thriftFactTable =
         fromWrapperToExternalTableSchema(wrapperTableInfo.getFactTable());
     return new org.apache.carbondata.format.TableInfo(thriftFactTable, new ArrayList<>());

core/src/main/java/org/apache/carbondata/core/util/CarbonUtil.java

@@ -50,6 +50,7 @@
 import java.util.UUID;
 
 import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.catalog.CatalogFactory;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.TableSpec;
 import org.apache.carbondata.core.datastore.block.AbstractIndex;
@@ -101,7 +102,6 @@
 import org.apache.carbondata.core.mutate.UpdateVO;
 import org.apache.carbondata.core.reader.CarbonHeaderReader;
 import org.apache.carbondata.core.reader.CarbonIndexFileReader;
-import org.apache.carbondata.core.reader.ThriftReader;
 import org.apache.carbondata.core.reader.ThriftReader.TBaseCreator;
 import org.apache.carbondata.core.scan.model.ProjectionDimension;
 import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
@@ -2015,13 +2015,7 @@ public static org.apache.carbondata.format.TableInfo readSchemaFile(String schem
   public static org.apache.carbondata.format.TableInfo readSchemaFile(String schemaFilePath,
       Configuration conf)
       throws IOException {
-    TBaseCreator createTBase = org.apache.carbondata.format.TableInfo::new;
-    ThriftReader thriftReader = new ThriftReader(schemaFilePath, createTBase, conf);
-    thriftReader.open();
-    org.apache.carbondata.format.TableInfo tableInfo =
-        (org.apache.carbondata.format.TableInfo) thriftReader.read();
-    thriftReader.close();
-    return tableInfo;
+    return CatalogFactory.getInstance().getCatalog().getSchema(schemaFilePath);
   }
 
   public static ColumnSchema thriftColumnSchemaToWrapperColumnSchema(

core/src/test/java/org/apache/carbondata/core/metadata/converter/ThriftWrapperSchemaConverterImplTest.java

@@ -1425,7 +1425,7 @@ public List<org.apache.carbondata.format.SchemaEvolutionEntry> getSchema_evoluti
     org.apache.carbondata.format.TableSchema thriftFactTable =
         new org.apache.carbondata.format.TableSchema("tableId", thriftColumnSchemas, schemaEvol);
     org.apache.carbondata.format.TableInfo actualResult = thriftWrapperSchemaConverter
-        .fromWrapperToExternalTableInfo(wrapperTableInfo, dbName, tableName);
+        .fromWrapperToExternalTableInfo(wrapperTableInfo);
     org.apache.carbondata.format.TableInfo expectedResult =
         new org.apache.carbondata.format.TableInfo(thriftFactTable, new ArrayList<org.apache
             .carbondata.format.TableSchema>());

core/src/test/java/org/apache/carbondata/core/metadata/schema/table/TableInfoTest.java

@@ -93,8 +93,7 @@ private TableInfo getTableInfo(String databaseName, String tableName) {
     CarbonTable carbonTable = CarbonTable.buildFromTableInfo(tableInfo);
     ThriftWrapperSchemaConverterImpl schemaConverter = new ThriftWrapperSchemaConverterImpl();
     org.apache.carbondata.format.TableInfo thriftTable = schemaConverter
-        .fromWrapperToExternalTableInfo(carbonTable.getTableInfo(), carbonTable.getDatabaseName(),
-            carbonTable.getTableName());
+        .fromWrapperToExternalTableInfo(carbonTable.getTableInfo());
     assertTrue(null != thriftTable);
   }
 
@@ -135,8 +134,7 @@ private TableInfo getTableInfo(String databaseName, String tableName) {
     CarbonTable carbonTable = CarbonTable.buildFromTableInfo(tableInfo);
     ThriftWrapperSchemaConverterImpl schemaConverter = new ThriftWrapperSchemaConverterImpl();
     org.apache.carbondata.format.TableInfo thriftTable = schemaConverter
-        .fromWrapperToExternalTableInfo(carbonTable.getTableInfo(), carbonTable.getDatabaseName(),
-            carbonTable.getTableName());
+        .fromWrapperToExternalTableInfo(carbonTable.getTableInfo());
     assertTrue(null != thriftTable);
   }
 

examples/spark/pom.xml

@@ -198,15 +198,6 @@
         <maven.test.skip>true</maven.test.skip>
       </properties>
     </profile>
-    <profile>
-      <id>spark-2.3</id>
-      <activation>
-        <activeByDefault>true</activeByDefault>
-      </activation>
-      <properties>
-        <spark.binary.version>2.3</spark.binary.version>
-      </properties>
-    </profile>
     <profile>
       <id>spark-2.4</id>
       <properties>
@@ -215,6 +206,9 @@
     </profile>
     <profile>
       <id>spark-3.1</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+      </activation>
       <properties>
         <spark.binary.version>3.1</spark.binary.version>
         <dep.jackson.version>2.10.0</dep.jackson.version>

hadoop/src/main/java/org/apache/carbondata/hadoop/testutil/StoreCreator.java

@@ -228,10 +228,7 @@ public CarbonTable createTable(
 
     SchemaConverter schemaConverter = new ThriftWrapperSchemaConverterImpl();
     org.apache.carbondata.format.TableInfo thriftTableInfo =
-        schemaConverter.fromWrapperToExternalTableInfo(
-            tableInfo,
-            tableInfo.getDatabaseName(),
-            tableInfo.getFactTable().getTableName());
+        schemaConverter.fromWrapperToExternalTableInfo(tableInfo);
     org.apache.carbondata.format.SchemaEvolutionEntry schemaEvolutionEntry =
         new org.apache.carbondata.format.SchemaEvolutionEntry(tableInfo.getLastUpdatedTime());
     thriftTableInfo.getFact_table().getSchema_evolution().getSchema_evolution_history()

index/examples/pom.xml

@@ -79,12 +79,6 @@
   </build>
 
   <profiles>
-    <profile>
-      <id>spark-2.3</id>
-      <properties>
-        <spark.binary.version>2.3</spark.binary.version>
-      </properties>
-    </profile>
     <profile>
       <id>spark-2.4</id>
       <properties>

index/secondary-index/pom.xml

@@ -156,15 +156,6 @@
         <maven.test.skip>true</maven.test.skip>
       </properties>
     </profile>
-    <profile>
-      <id>spark-2.3</id>
-      <activation>
-        <activeByDefault>true</activeByDefault>
-      </activation>
-      <properties>
-        <spark.binary.version>2.3</spark.binary.version>
-      </properties>
-    </profile>
     <profile>
       <id>spark-2.4</id>
       <properties>
@@ -173,6 +164,9 @@
     </profile>
     <profile>
       <id>spark-3.1</id>
+      <activation>
+        <activeByDefault>true</activeByDefault>
+      </activation>
       <properties>
         <spark.binary.version>3.1</spark.binary.version>
       </properties>

integration/flink/pom.xml

@@ -218,29 +218,6 @@
     </dependencies>
 
     <profiles>
-        <profile>
-            <id>spark-2.3</id>
-            <activation>
-                <activeByDefault>true</activeByDefault>
-            </activation>
-            <properties>
-                <spark.binary.version>2.3</spark.binary.version>
-            </properties>
-            <dependencies>
-                <dependency>
-                    <groupId>org.apache.carbondata</groupId>
-                    <artifactId>carbondata-spark_${spark.binary.version}</artifactId>
-                    <version>${project.version}</version>
-                    <scope>test</scope>
-                    <exclusions>
-                        <exclusion>
-                            <groupId>org.apache.hive</groupId>
-                            <artifactId>hive-exec</artifactId>
-                        </exclusion>
-                    </exclusions>
-                </dependency>
-            </dependencies>
-        </profile>
         <profile>
             <id>spark-2.4</id>
             <properties>
@@ -263,6 +240,9 @@
         </profile>
       <profile>
         <id>spark-3.1</id>
+          <activation>
+              <activeByDefault>true</activeByDefault>
+          </activation>
         <properties>
           <spark.binary.version>3.1</spark.binary.version>
         </properties>

integration/hive/src/main/java/org/apache/carbondata/hive/util/HiveCarbonUtil.java

@@ -255,8 +255,7 @@ private static void writeSchemaFile(TableInfo tableInfo) throws IOException {
     ThriftWriter thriftWriter = new ThriftWriter(schemaFilePath, false);
     thriftWriter.open(FileWriteOperation.OVERWRITE);
     thriftWriter.write(schemaConverter
-        .fromWrapperToExternalTableInfo(tableInfo, tableInfo.getDatabaseName(),
-            tableInfo.getFactTable().getTableName()));
+        .fromWrapperToExternalTableInfo(tableInfo));
     thriftWriter.close();
   }
 

integration/presto/src/main/prestosql/org/apache/carbondata/presto/impl/CarbonTableReader.java

@@ -21,16 +21,16 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
-import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 import org.apache.carbondata.common.logging.LogServiceFactory;
+import org.apache.carbondata.core.catalog.BaseCatalog;
+import org.apache.carbondata.core.catalog.CatalogFactory;
 import org.apache.carbondata.core.constants.CarbonCommonConstants;
 import org.apache.carbondata.core.datastore.filesystem.CarbonFile;
 import org.apache.carbondata.core.datastore.impl.FileFactory;
@@ -40,38 +40,30 @@
 import org.apache.carbondata.core.indexstore.PartitionSpec;
 import org.apache.carbondata.core.metadata.AbsoluteTableIdentifier;
 import org.apache.carbondata.core.metadata.CarbonMetadata;
-import org.apache.carbondata.core.metadata.SegmentFileStore;
 import org.apache.carbondata.core.metadata.converter.SchemaConverter;
 import org.apache.carbondata.core.metadata.converter.ThriftWrapperSchemaConverterImpl;
 import org.apache.carbondata.core.metadata.index.IndexType;
 import org.apache.carbondata.core.metadata.schema.indextable.IndexMetadata;
 import org.apache.carbondata.core.metadata.schema.indextable.IndexTableInfo;
 import org.apache.carbondata.core.metadata.schema.table.CarbonTable;
 import org.apache.carbondata.core.metadata.schema.table.TableInfo;
-import org.apache.carbondata.core.reader.ThriftReader;
 import org.apache.carbondata.core.scan.expression.Expression;
 import org.apache.carbondata.core.statusmanager.FileFormat;
-import org.apache.carbondata.core.statusmanager.LoadMetadataDetails;
 import org.apache.carbondata.core.util.CarbonProperties;
 import org.apache.carbondata.core.util.CarbonUtil;
 import org.apache.carbondata.core.util.path.CarbonTablePath;
 import org.apache.carbondata.hadoop.CarbonInputSplit;
 import org.apache.carbondata.hadoop.api.CarbonInputFormat;
 import org.apache.carbondata.hadoop.api.CarbonTableInputFormat;
-import org.apache.carbondata.presto.PrestoFilterUtil;
 
 import com.google.gson.Gson;
 import com.google.inject.Inject;
-import io.prestosql.plugin.hive.HiveColumnHandle;
 import io.prestosql.spi.connector.SchemaTableName;
-import io.prestosql.spi.predicate.TupleDomain;
-import org.apache.commons.collections.CollectionUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.mapred.JobConf;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.log4j.Logger;
-import org.apache.thrift.TBase;
 
 import static org.apache.hadoop.fs.s3a.Constants.ACCESS_KEY;
 import static org.apache.hadoop.fs.s3a.Constants.ENDPOINT;
@@ -192,20 +184,7 @@ private CarbonTableCacheModel parseCarbonMetadata(SchemaTableName table, String
         org.apache.carbondata.format.TableInfo tableInfo;
         long modifiedTime = System.currentTimeMillis();
         if (isTransactionalTable) {
-          //Step 2: read the metadata (tableInfo) of the table.
-          ThriftReader.TBaseCreator createTBase = new ThriftReader.TBaseCreator() {
-            // TBase is used to read and write thrift objects.
-            // TableInfo is a kind of TBase used to read and write table information.
-            // TableInfo is generated by thrift,
-            // see schema.thrift under format/src/main/thrift for details.
-            public TBase create() {
-              return new org.apache.carbondata.format.TableInfo();
-            }
-          };
-          ThriftReader thriftReader = new ThriftReader(schemaFilePath, createTBase, config);
-          thriftReader.open();
-          tableInfo = (org.apache.carbondata.format.TableInfo) thriftReader.read();
-          thriftReader.close();
+          tableInfo = CatalogFactory.getInstance().getCatalog().getSchema(schemaFilePath);
           modifiedTime = schemaFile.getLastModifiedTime();
         } else {
           tableInfo = CarbonUtil.inferSchema(tablePath, table.getTableName(), false, config);

integration/presto/src/test/scala/org/apache/carbondata/presto/util/CarbonDataStoreCreator.scala

@@ -153,10 +153,7 @@ object CarbonDataStoreCreator {
     val schemaConverter: SchemaConverter =
       new ThriftWrapperSchemaConverterImpl()
     val thriftTableInfo: TableInfo =
-      schemaConverter.fromWrapperToExternalTableInfo(
-        tableInfo,
-        tableInfo.getDatabaseName,
-        tableInfo.getFactTable.getTableName)
+      schemaConverter.fromWrapperToExternalTableInfo(tableInfo)
     val schemaEvolutionEntry: SchemaEvolutionEntry =
       new org.apache.carbondata.format.SchemaEvolutionEntry(
         tableInfo.getLastUpdatedTime)