DRILL-7058: Refresh command to support subset of columns

closes #1666

7 files changed, 160 insertions, 42 deletions

diff --git a/exec/java-exec/src/main/codegen/data/Parser.tdd b/exec/java-exec/src/main/codegen/data/Parser.tdd
index 4461f6d98..358038853 100644
--- a/exec/java-exec/src/main/codegen/data/Parser.tdd
+++ b/exec/java-exec/src/main/codegen/data/Parser.tdd
@@ -37,12 +37,13 @@
     "METADATA",
     "IF",
     "JAR",
-    "PROPERTIES"
+    "PROPERTIES",
     "ANALYZE",
     "COMPUTE",
     "ESTIMATE",
     "STATISTICS",
-    "SAMPLE"
+    "SAMPLE",
+    "COLUMNS"
   ]
 
   # List of methods for parsing custom SQL statements.
@@ -854,6 +855,7 @@
         "YEAR",
         # "YEARS", # not a keyword in Calcite
         "ZONE",
+        "COLUMNS"
       ]
 
   # List of additional join types. Each is a method with no arguments.
diff --git a/exec/java-exec/src/main/codegen/includes/parserImpls.ftl b/exec/java-exec/src/main/codegen/includes/parserImpls.ftl
index 2da1c4a64..015ba9452 100644
--- a/exec/java-exec/src/main/codegen/includes/parserImpls.ftl
+++ b/exec/java-exec/src/main/codegen/includes/parserImpls.ftl
@@ -464,22 +464,30 @@ SqlNode SqlDropSchema(SqlParserPos pos) :
 
 /**
  * Parse refresh table metadata statement.
- * REFRESH TABLE METADATA tblname
+ * REFRESH TABLE METADATA [COLUMNS ((field1, field2,..) | NONE)] tblname
  */
 SqlNode SqlRefreshMetadata() :
 {
     SqlParserPos pos;
     SqlIdentifier tblName;
-    SqlNodeList fieldList;
+    SqlNodeList fieldList = null;
     SqlNode query;
+    boolean allColumns = true;
 }
 {
     <REFRESH> { pos = getPos(); }
     <TABLE>
     <METADATA>
+    [
+        <COLUMNS> { allColumns = false; }
+        (   fieldList = ParseRequiredFieldList("Table")
+            |
+            <NONE>
+        )
+    ]
     tblName = CompoundIdentifier()
     {
-        return new SqlRefreshMetadata(pos, tblName);
+        return new SqlRefreshMetadata(pos, tblName, SqlLiteral.createBoolean(allColumns, getPos()), fieldList);
     }
 }
 
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java
index 3ca778706..0a02f038b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/RefreshMetadataHandler.java
@@ -17,6 +17,13 @@
  */
 package org.apache.drill.exec.planner.sql.handlers;
 
+import java.util.HashSet;
+import java.util.Set;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.drill.common.expression.SchemaPath;
 import static org.apache.drill.exec.planner.sql.SchemaUtilites.findSchema;
 
 import org.apache.calcite.schema.SchemaPlus;
@@ -69,6 +76,9 @@ public class RefreshMetadataHandler extends DefaultSqlHandler {
       }
 
       final String tableName = refreshTable.getName();
+      final SqlNodeList columnList = getColumnList(refreshTable);
+      final Set<String> columnSet = getColumnRootSegments(columnList);
+      final SqlLiteral allColumns = refreshTable.getAllColumns();
 
       if (tableName.contains("*") || tableName.contains("?")) {
         return direct(false, "Glob path %s not supported for metadata refresh", tableName);
@@ -121,7 +131,7 @@ public class RefreshMetadataHandler extends DefaultSqlHandler {
         .withFormatConfig((ParquetFormatConfig) formatConfig)
         .withOptions(context.getOptions())
         .build();
-      Metadata.createMeta(fs, selectionRoot, readerConfig);
+      Metadata.createMeta(fs, selectionRoot, readerConfig, allColumns.booleanValue(), columnSet);
       return direct(true, "Successfully updated metadata for table %s.", tableName);
 
     } catch(Exception e) {
@@ -130,5 +140,29 @@ public class RefreshMetadataHandler extends DefaultSqlHandler {
     }
   }
 
+  private Set<String> getColumnRootSegments(SqlNodeList columnList) {
+    Set<String> columnSet = new HashSet<>();
+    if (columnList != null) {
+      for (SqlNode column : columnList.getList()) {
+        // Add only the root segment. Collect metadata for all the columns under that root segment
+        columnSet.add(SchemaPath.parseFromString(column.toString()).getRootSegmentPath());
+      }
+    }
+    return columnSet;
+  }
+
+  /**
+   * Generates the column list specified in the Refresh statement
+   * @param sqlrefreshMetadata sql parse node representing refresh statement
+   * @return list of columns specified in the refresh command
+   */
+  private SqlNodeList getColumnList(final SqlRefreshMetadata sqlrefreshMetadata) {
+    SqlNodeList columnList = sqlrefreshMetadata.getFieldList();
+    if (columnList == null || !SqlNodeList.isEmptyList(columnList)) {
+      columnList = new SqlNodeList(SqlParserPos.ZERO);
+      columnList.add(new SqlIdentifier(SchemaPath.STAR_COLUMN.rootName(), SqlParserPos.ZERO));
+    }
+    return columnList;
+  }
 
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java
index 56985e2fe..5d3d08022 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/CompoundIdentifierConverter.java
@@ -76,7 +76,7 @@ public class CompoundIdentifierConverter extends SqlShuttle {
         .put(SqlJoin.class, arrayOf(D, D, D, D, D, E))
         .put(SqlOrderBy.class, arrayOf(D, E, D, D))
         .put(SqlDropTable.class, arrayOf(D, D))
-        .put(SqlRefreshMetadata.class, arrayOf(D))
+        .put(SqlRefreshMetadata.class, arrayOf(D, D, E))
         .put(SqlSetOption.class, arrayOf(D, D, D))
         .put(SqlCreateFunction.class, arrayOf(D))
         .put(SqlDropFunction.class, arrayOf(D))
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlRefreshMetadata.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlRefreshMetadata.java
index 84d95f4e0..352357e90 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlRefreshMetadata.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/parser/SqlRefreshMetadata.java
@@ -24,6 +24,7 @@ import org.apache.calcite.sql.SqlIdentifier;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlLiteral;
 import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.SqlSpecialOperator;
 import org.apache.calcite.sql.SqlWriter;
@@ -43,15 +44,19 @@ public class SqlRefreshMetadata extends DrillSqlCall {
   public static final SqlSpecialOperator OPERATOR = new SqlSpecialOperator("REFRESH_TABLE_METADATA", SqlKind.OTHER_DDL) {
     @Override
     public SqlCall createCall(SqlLiteral functionQualifier, SqlParserPos pos, SqlNode... operands) {
-      return new SqlRefreshMetadata(pos, (SqlIdentifier) operands[0]);
+      return new SqlRefreshMetadata(pos, (SqlIdentifier) operands[0], (SqlLiteral) operands[1], (SqlNodeList) operands[2]);
     }
   };
 
   private SqlIdentifier tblName;
+  private final SqlLiteral allColumns;
+  private final SqlNodeList fieldList;
 
-  public SqlRefreshMetadata(SqlParserPos pos, SqlIdentifier tblName){
+  public SqlRefreshMetadata(SqlParserPos pos, SqlIdentifier tblName, SqlLiteral allColumns, SqlNodeList fieldList){
     super(pos);
     this.tblName = tblName;
+    this.allColumns = allColumns;
+    this.fieldList = fieldList;
   }
 
   @Override
@@ -63,6 +68,8 @@ public class SqlRefreshMetadata extends DrillSqlCall {
   public List<SqlNode> getOperandList() {
     List<SqlNode> ops = Lists.newArrayList();
     ops.add(tblName);
+    ops.add(allColumns);
+    ops.add(fieldList);
     return ops;
   }
 
@@ -71,6 +78,20 @@ public class SqlRefreshMetadata extends DrillSqlCall {
     writer.keyword("REFRESH");
     writer.keyword("TABLE");
     writer.keyword("METADATA");
+    if (!allColumns.booleanValue()) {
+      writer.keyword("COLUMNS");
+      if (fieldList == null) {
+        writer.keyword("NONE");
+      } else if (fieldList != null && fieldList.size() > 0) {
+        writer.keyword("(");
+        fieldList.get(0).unparse(writer, leftPrec, rightPrec);
+        for (int i = 1; i < fieldList.size(); i++) {
+          writer.keyword(",");
+          fieldList.get(i).unparse(writer, leftPrec, rightPrec);
+        }
+        writer.keyword(")");
+      }
+    }
     tblName.unparse(writer, leftPrec, rightPrec);
   }
 
@@ -94,4 +115,12 @@ public class SqlRefreshMetadata extends DrillSqlCall {
   public AbstractSqlHandler getSqlHandler(SqlHandlerConfig config) {
     return new RefreshMetadataHandler(config);
   }
+
+  public SqlNodeList getFieldList() {
+    return fieldList;
+  }
+
+  public SqlLiteral getAllColumns() {
+    return allColumns;
+  }
 }
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
index d0e2734d2..18cbb6627 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
@@ -25,6 +25,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.module.SimpleModule;
 import com.fasterxml.jackson.module.afterburner.AfterburnerModule;
 import org.apache.drill.exec.serialization.PathSerDe;
+import java.util.Set;
 import org.apache.drill.exec.store.parquet.ParquetReaderConfig;
 import org.apache.drill.shaded.guava.com.google.common.base.Stopwatch;
 import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
@@ -105,14 +106,15 @@ public class Metadata {
 
   /**
    * Create the parquet metadata file for the directory at the given path, and for any subdirectories.
-   *
    * @param fs file system
    * @param path path
    * @param readerConfig parquet reader configuration
+   * @param allColumns if set, store column metadata for all the columns
+   * @param columnSet Set of columns for which column metadata has to be stored
    */
-  public static void createMeta(FileSystem fs, Path path, ParquetReaderConfig readerConfig) throws IOException {
+  public static void createMeta(FileSystem fs, Path path, ParquetReaderConfig readerConfig, boolean allColumns, Set<String> columnSet) throws IOException {
     Metadata metadata = new Metadata(readerConfig);
-    metadata.createMetaFilesRecursively(path, fs);
+    metadata.createMetaFilesRecursively(path, fs, allColumns, columnSet);
   }
 
   /**
@@ -207,12 +209,14 @@ public class Metadata {
    *
    * @param path to the directory of the parquet table
    * @param fs file system
+   * @param allColumns if set, store column metadata for all the columns
+   * @param columnSet Set of columns for which column metadata has to be stored
    * @return Pair of parquet metadata. The left one is a parquet metadata for the table. The right one of the Pair is
    *         a metadata for all subdirectories (if they are present and there are no any parquet files in the
    *         {@code path} directory).
    * @throws IOException if parquet metadata can't be serialized and written to the json file
    */
-  private Pair<ParquetTableMetadata_v3, ParquetTableMetadataDirs> createMetaFilesRecursively(final Path path, FileSystem fs) throws IOException {
+  private Pair<ParquetTableMetadata_v3, ParquetTableMetadataDirs> createMetaFilesRecursively(final Path path, FileSystem fs, boolean allColumns, Set<String> columnSet) throws IOException {
     Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
     List<ParquetFileMetadata_v3> metaDataList = Lists.newArrayList();
     List<Path> directoryList = Lists.newArrayList();
@@ -226,7 +230,7 @@ public class Metadata {
 
     for (final FileStatus file : DrillFileSystemUtil.listAll(fs, p, false)) {
       if (file.isDirectory()) {
-        ParquetTableMetadata_v3 subTableMetadata = (createMetaFilesRecursively(file.getPath(), fs)).getLeft();
+        ParquetTableMetadata_v3 subTableMetadata = (createMetaFilesRecursively(file.getPath(), fs, allColumns, columnSet)).getLeft();
         metaDataList.addAll(subTableMetadata.files);
         directoryList.addAll(subTableMetadata.directories);
         directoryList.add(file.getPath());
@@ -240,7 +244,7 @@ public class Metadata {
     ParquetTableMetadata_v3 parquetTableMetadata = new ParquetTableMetadata_v3(SUPPORTED_VERSIONS.last().toString(),
                                                                                 DrillVersionInfo.getVersion());
     if (childFiles.size() > 0) {
-      List<ParquetFileMetadata_v3 > childFilesMetadata = getParquetFileMetadata_v3(parquetTableMetadata, childFiles);
+      List<ParquetFileMetadata_v3 > childFilesMetadata = getParquetFileMetadata_v3(parquetTableMetadata, childFiles, allColumns, columnSet);
       metaDataList.addAll(childFilesMetadata);
       // Note that we do not need to merge the columnInfo at this point. The columnInfo is already added
       // to the parquetTableMetadata.
@@ -330,7 +334,7 @@ public class Metadata {
       throws IOException {
     ParquetTableMetadata_v3 tableMetadata = new ParquetTableMetadata_v3(SUPPORTED_VERSIONS.last().toString(),
                                                                         DrillVersionInfo.getVersion());
-    tableMetadata.files = getParquetFileMetadata_v3(tableMetadata, fileStatusMap);
+    tableMetadata.files = getParquetFileMetadata_v3(tableMetadata, fileStatusMap, true, null);
     tableMetadata.directories = new ArrayList<>();
     return tableMetadata;
   }
@@ -341,14 +345,15 @@ public class Metadata {
    * @param parquetTableMetadata_v3 can store column schema info from all the files and row groups
    * @param fileStatusMap parquet files statuses and corresponding file systems
    *
+   * @param allColumns if set, store column metadata for all the columns
+   * @param columnSet Set of columns for which column metadata has to be stored
    * @return list of the parquet file metadata with absolute paths
    * @throws IOException is thrown in case of issues while executing the list of runnables
    */
-  private List<ParquetFileMetadata_v3> getParquetFileMetadata_v3(
-      ParquetTableMetadata_v3 parquetTableMetadata_v3, Map<FileStatus, FileSystem> fileStatusMap) throws IOException {
+  private List<ParquetFileMetadata_v3> getParquetFileMetadata_v3(ParquetTableMetadata_v3 parquetTableMetadata_v3, Map<FileStatus, FileSystem> fileStatusMap, boolean allColumns, Set<String> columnSet) throws IOException {
     return TimedCallable.run("Fetch parquet metadata", logger,
         Collectors.toList(fileStatusMap,
-            (fileStatus, fileSystem) -> new MetadataGatherer(parquetTableMetadata_v3, fileStatus, fileSystem)),
+            (fileStatus, fileSystem) -> new MetadataGatherer(parquetTableMetadata_v3, fileStatus, fileSystem, allColumns, columnSet)),
         16
     );
   }
@@ -361,16 +366,20 @@ public class Metadata {
     private final ParquetTableMetadata_v3 parquetTableMetadata;
     private final FileStatus fileStatus;
     private final FileSystem fs;
+    private final boolean allColumns;
+    private final Set<String> columnSet;
 
-    MetadataGatherer(ParquetTableMetadata_v3 parquetTableMetadata, FileStatus fileStatus, FileSystem fs) {
+    MetadataGatherer(ParquetTableMetadata_v3 parquetTableMetadata, FileStatus fileStatus, FileSystem fs, boolean allColumns, Set<String> columnSet) {
       this.parquetTableMetadata = parquetTableMetadata;
       this.fileStatus = fileStatus;
       this.fs = fs;
+      this.allColumns = allColumns;
+      this.columnSet = columnSet;
     }
 
     @Override
     protected ParquetFileMetadata_v3 runInner() throws Exception {
-      return getParquetFileMetadata_v3(parquetTableMetadata, fileStatus, fs);
+      return getParquetFileMetadata_v3(parquetTableMetadata, fileStatus, fs, allColumns, columnSet);
     }
 
     public String toString() {
@@ -417,7 +426,7 @@ public class Metadata {
    * Get the metadata for a single file
    */
   private ParquetFileMetadata_v3 getParquetFileMetadata_v3(ParquetTableMetadata_v3 parquetTableMetadata,
-      final FileStatus file, final FileSystem fs) throws IOException, InterruptedException {
+     final FileStatus file, final FileSystem fs, boolean allColumns, Set<String> columnSet) throws IOException, InterruptedException {
     final ParquetMetadata metadata;
     final UserGroupInformation processUserUgi = ImpersonationUtil.getProcessUserUGI();
     final Configuration conf = new Configuration(fs.getConf());
@@ -453,7 +462,6 @@ public class Metadata {
       List<ColumnMetadata_v3> columnMetadataList = new ArrayList<>();
       long length = 0;
       for (ColumnChunkMetaData col : rowGroup.getColumns()) {
-        Statistics<?> stats = col.getStatistics();
         String[] columnName = col.getPath().toArray();
         SchemaPath columnSchemaName = SchemaPath.getCompoundPath(columnName);
         ColTypeInfo colTypeInfo = colTypeInfoMap.get(columnSchemaName);
@@ -466,25 +474,28 @@ public class Metadata {
           parquetTableMetadata.columnTypeInfo = new ConcurrentHashMap<>();
         }
         parquetTableMetadata.columnTypeInfo.put(new ColumnTypeMetadata_v3.Key(columnTypeMetadata.name), columnTypeMetadata);
-
-        // Save the column schema info. We'll merge it into one list
-        Object minValue = null;
-        Object maxValue = null;
-        long numNulls = -1;
-        boolean statsAvailable = stats != null && !stats.isEmpty();
-        if (statsAvailable) {
-          if (stats.hasNonNullValue()) {
-            minValue = stats.genericGetMin();
-            maxValue = stats.genericGetMax();
-            if (containsCorruptDates == ParquetReaderUtility.DateCorruptionStatus.META_SHOWS_CORRUPTION && columnTypeMetadata.originalType == OriginalType.DATE) {
-              minValue = ParquetReaderUtility.autoCorrectCorruptedDate((Integer) minValue);
-              maxValue = ParquetReaderUtility.autoCorrectCorruptedDate((Integer) maxValue);
+        // Store column metadata only if allColumns is set to true or if the column belongs to the subset of columns specified in the refresh command
+        if (allColumns || columnSet == null || !allColumns && columnSet != null && columnSet.size() > 0 && columnSet.contains(columnSchemaName.getRootSegmentPath())) {
+          Statistics<?> stats = col.getStatistics();
+          // Save the column schema info. We'll merge it into one list
+          Object minValue = null;
+          Object maxValue = null;
+          long numNulls = -1;
+          boolean statsAvailable = stats != null && !stats.isEmpty();
+          if (statsAvailable) {
+            if (stats.hasNonNullValue()) {
+              minValue = stats.genericGetMin();
+              maxValue = stats.genericGetMax();
+              if (containsCorruptDates == ParquetReaderUtility.DateCorruptionStatus.META_SHOWS_CORRUPTION && columnTypeMetadata.originalType == OriginalType.DATE) {
+                minValue = ParquetReaderUtility.autoCorrectCorruptedDate((Integer) minValue);
+                maxValue = ParquetReaderUtility.autoCorrectCorruptedDate((Integer) maxValue);
+              }
             }
+            numNulls = stats.getNumNulls();
           }
-          numNulls = stats.getNumNulls();
+          ColumnMetadata_v3 columnMetadata = new ColumnMetadata_v3(columnTypeMetadata.name, col.getPrimitiveType().getPrimitiveTypeName(), minValue, maxValue, numNulls);
+          columnMetadataList.add(columnMetadata);
         }
-        ColumnMetadata_v3 columnMetadata = new ColumnMetadata_v3(columnTypeMetadata.name, col.getPrimitiveType().getPrimitiveTypeName(), minValue, maxValue, numNulls);
-        columnMetadataList.add(columnMetadata);
         length += col.getTotalSize();
       }
 
@@ -610,7 +621,7 @@ public class Metadata {
         parquetTableMetadataDirs.updateRelativePaths(metadataParentDirPath);
         if (!alreadyCheckedModification && tableModified(parquetTableMetadataDirs.getDirectories(), path, metadataParentDir, metaContext, fs)) {
           parquetTableMetadataDirs =
-              (createMetaFilesRecursively(Path.getPathWithoutSchemeAndAuthority(path.getParent()), fs)).getRight();
+              (createMetaFilesRecursively(Path.getPathWithoutSchemeAndAuthority(path.getParent()), fs, true, null)).getRight();
           newMetadata = true;
         }
       } else {
@@ -622,9 +633,10 @@ public class Metadata {
         if (new MetadataVersion(parquetTableMetadata.getMetadataVersion()).compareTo(new MetadataVersion(3, 0)) >= 0) {
           ((ParquetTableMetadata_v3) parquetTableMetadata).updateRelativePaths(metadataParentDirPath);
         }
-        if (!alreadyCheckedModification && tableModified(parquetTableMetadata.getDirectories(), path, metadataParentDir, metaContext, fs)) {
+          if (!alreadyCheckedModification && tableModified(parquetTableMetadata.getDirectories(), path, metadataParentDir, metaContext, fs)) {
+          // TODO change with current columns in existing metadata (auto refresh feature)
           parquetTableMetadata =
-              (createMetaFilesRecursively(Path.getPathWithoutSchemeAndAuthority(path.getParent()), fs)).getLeft();
+              (createMetaFilesRecursively(Path.getPathWithoutSchemeAndAuthority(path.getParent()), fs, true, null)).getLeft();
           newMetadata = true;
         }
 
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java
index f85ef2b71..3500f6f30 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/parquet/TestParquetMetadataCache.java
@@ -924,6 +924,39 @@ public class TestParquetMetadataCache extends PlanTestBase {
     // TODO: Check that metadata cache file is actually regenerated, once Drill will use JDK version with resolved JDK-8177809.
   }
 
+  @Test
+  public void testRefreshDefault() throws Exception {
+    test("refresh table metadata dfs.`%s`", TABLE_NAME_1);
+    checkForMetadataFile(TABLE_NAME_1);
+    String query = String.format("select dir0, dir1, o_custkey, o_orderdate from dfs.`%s` " +
+            " where dir0=1994 and dir1 in ('Q1', 'Q2')", TABLE_NAME_1);
+    int expectedRowCount = 20;
+    int actualRowCount = testSql(query);
+    assertEquals(expectedRowCount, actualRowCount);
+  }
+
+  @Test
+  public void testRefreshWithColumns() throws Exception {
+    test("refresh table metadata columns (o_custkey, o_orderdate) dfs.`%s`", TABLE_NAME_1);
+    checkForMetadataFile(TABLE_NAME_1);
+    String query = String.format("select dir0, dir1, o_custkey, o_orderdate from dfs.`%s` " +
+            " where dir0=1994 and dir1 in ('Q1', 'Q2')", TABLE_NAME_1);
+    int expectedRowCount = 20;
+    int actualRowCount = testSql(query);
+    assertEquals(expectedRowCount, actualRowCount);
+  }
+
+  @Test
+  public void testRefreshNone() throws Exception {
+    test("refresh table metadata columns none dfs.`%s`", TABLE_NAME_1);
+    checkForMetadataFile(TABLE_NAME_1);
+    String query = String.format("select dir0, dir1, o_custkey, o_orderdate from dfs.`%s` " +
+            " where dir0=1994 and dir1 in ('Q1', 'Q2')", TABLE_NAME_1);
+    int expectedRowCount = 20;
+    int actualRowCount = testSql(query);
+    assertEquals(expectedRowCount, actualRowCount);
+  }
+
   /**
    * Helper method for checking the metadata file existence
    *