using same column reference across table and store descriptor

addons/hive-bridge/src/main/java/org/apache/hadoop/metadata/hive/bridge/HiveMetaStoreBridge.java

@@ -271,9 +271,12 @@ public class HiveMetaStoreBridge {
             // add reference to the database
             tableRef.set("dbName", dbReference);
 
+            List<Referenceable> colList = getColumns(hiveTable.getCols());
+            tableRef.set("columns", colList);
+
             // add reference to the StorageDescriptor
             StorageDescriptor storageDesc = hiveTable.getSd();
-            Referenceable sdReferenceable = fillStorageDescStruct(storageDesc);
+            Referenceable sdReferenceable = fillStorageDescStruct(storageDesc, colList);
             tableRef.set("sd", sdReferenceable);
 
             // add reference to the Partition Keys
@@ -293,8 +296,6 @@ public class HiveMetaStoreBridge {
             tableRef.set("tableType", hiveTable.getTableType().name());
             tableRef.set("temporary", hiveTable.isTemporary());
 
-            List<Referenceable> colList = getColumns(hiveTable.getAllCols());
-            tableRef.set("columns", colList);
 
             tableRef = createInstance(tableRef);
         } else {
@@ -388,7 +389,7 @@ public class HiveMetaStoreBridge {
         indexRef.set("origTableName", index.getOrigTableName());
         indexRef.set("indexTableName", index.getIndexTableName());
 
-        Referenceable sdReferenceable = fillStorageDescStruct(index.getSd());
+        Referenceable sdReferenceable = fillStorageDescStruct(index.getSd(), null);
         indexRef.set("sd", sdReferenceable);
 
         indexRef.set("parameters", index.getParameters());
@@ -398,7 +399,7 @@ public class HiveMetaStoreBridge {
         createInstance(indexRef);
     }
 
-    private Referenceable fillStorageDescStruct(StorageDescriptor storageDesc) throws Exception {
+    private Referenceable fillStorageDescStruct(StorageDescriptor storageDesc, List<Referenceable> colList) throws Exception {
         LOG.debug("Filling storage descriptor information for " + storageDesc);
 
         Referenceable sdReferenceable = new Referenceable(HiveDataTypes.HIVE_STORAGEDESC.getName());
@@ -433,8 +434,15 @@ public class HiveMetaStoreBridge {
         }
         */
 
-        List<Referenceable> fieldsList = getColumns(storageDesc.getCols());
-        sdReferenceable.set("cols", fieldsList);
+        //Use the passed column list if not null, ex: use same references for table and SD
+        List<FieldSchema> columns = storageDesc.getCols();
+        if (columns != null && !columns.isEmpty()) {
+            if (colList != null) {
+                sdReferenceable.set("cols", colList);
+            } else {
+                sdReferenceable.set("cols", getColumns(columns));
+            }
+        }
 
         List<Struct> sortColsStruct = new ArrayList<>();
         for (Order sortcol : storageDesc.getSortCols()) {

addons/hive-bridge/src/test/java/org/apache/hadoop/metadata/hive/hook/HiveHookIT.java

@@ -127,8 +127,11 @@ public class HiveHookIT {
     public void testCreateTable() throws Exception {
         String tableName = tableName();
         String dbName = createDatabase();
-        runCommand("create table " + dbName + "." + tableName + "(id int, name string)");
+        String colName = "col" + random();
+        runCommand("create table " + dbName + "." + tableName + "(" + colName + " int, name string)");
         assertTableIsRegistered(dbName, tableName);
+        //there is only one instance of column registered
+        assertColumnIsRegistered(colName);
 
         tableName = createTable();
         String tableId = assertTableIsRegistered(DEFAULT_DB, tableName);
@@ -139,6 +142,13 @@ public class HiveHookIT {
         assertDatabaseIsRegistered(DEFAULT_DB);
     }
 
+    private String assertColumnIsRegistered(String colName) throws Exception {
+        LOG.debug("Searching for column {}", colName);
+        String query = String.format("%s where name = '%s'", HiveDataTypes.HIVE_COLUMN.getName(), colName.toLowerCase());
+        return assertEntityIsRegistered(query, true);
+
+    }
+
     @Test
     public void testCTAS() throws Exception {
         String tableName = createTable();