[DRAFT] HIVE-29799: Table containing flattened union files cannot be converted to ACID

ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java

@@ -145,12 +145,18 @@ public void flattenUnionSubdirectories(Path sourcePath) throws HiveException {
           Path path = i.next().getPath();
           Path parent = path.getParent();
           if (parent.getName().startsWith(prefix)) {
-            // We do rename by including the name of parent directory into the filename so that there are no clashes
-            // when we move the files to the parent directory. Ex. HIVE_UNION_SUBDIR_1/000000_0 -> 1_000000_0
+            // Fold the subdir index into the filename so there are no clashes when the files
+            // are moved to the parent directory, and so that the resulting name still matches
+            // the "original data file" convention that both the ACID reader (AcidUtils.
+            // ORIGINAL_PATTERN_COPY) and the metastore's non-ACID→ACID validator (Transactional
+            // ValidationListener.ORIGINAL_PATTERN_COPY) recognize:
+            //   HIVE_UNION_SUBDIR_1/000000_0 -> 000000_0_copy_1
+            // See Utilities.COPY_KEYWORD.
             String parentOfParent = parent.getParent().toString();
             String parentNameSuffix = parent.getName().substring(prefix.length());
 
-            fs.rename(path, new Path(parentOfParent + "/" + parentNameSuffix + "_" + path.getName()));
+            fs.rename(path,
+                new Path(parentOfParent + "/" + path.getName() + Utilities.COPY_KEYWORD + parentNameSuffix));
 
             unionSubdirs.add(parent);
           }

ql/src/java/org/apache/hadoop/hive/ql/io/orc/VectorizedOrcAcidRowBatchReader.java

@@ -341,7 +341,8 @@ private VectorizedOrcAcidRowBatchReader(JobConf conf, OrcSplit orcSplit, Reporte
       if (deleteEventRegistry.isEmpty() && !rowIdProjected) {
         Path parent = orcSplit.getPath().getParent();
         while (parent != null && !rootPath.equals(parent)) {
-          if (parent.getName().startsWith(AcidUtils.BASE_PREFIX)) {
+          String parentName = parent.getName();
+          if (parentName.startsWith(AcidUtils.BASE_PREFIX)) {
             /**
              * The assumption here is that any base_x is filtered out by
              * {@link AcidUtils#getAcidState(Path, Configuration, ValidWriteIdList)}
@@ -351,7 +352,8 @@ private VectorizedOrcAcidRowBatchReader(JobConf conf, OrcSplit orcSplit, Reporte
              */
             readerOptions.includeAcidColumns(false);
             break;
-          } else {
+          } else if (parentName.startsWith(AcidUtils.DELTA_PREFIX)
+              || parentName.startsWith(AcidUtils.DELETE_DELTA_PREFIX)) {
             ParsedDeltaLight pd = ParsedDeltaLight.parse(parent);
             if (validWriteIdList.isWriteIdRangeValid(pd.getMinWriteId(),
                 pd.getMaxWriteId()) == ValidWriteIdList.RangeResponse.ALL) {
@@ -361,6 +363,11 @@ private VectorizedOrcAcidRowBatchReader(JobConf conf, OrcSplit orcSplit, Reporte
               break;
             }
           }
+          // Any other directory name (e.g. HIVE_UNION_SUBDIR_<N>/ produced by
+          // an INSERT ... UNION ALL against a table that was later converted
+          // to full ACID) is not an ACID container — skip it and keep walking
+          // up. Feeding such a name into ParsedDeltaLight.parse would throw
+          // NumberFormatException.
           parent = parent.getParent();
         }
       }

ql/src/test/org/apache/hadoop/hive/ql/exec/TestMoveTask.java

@@ -34,7 +34,9 @@ public class TestMoveTask {
   @Test
   public void flattenUnionSubdirectories() throws IOException, HiveException {
     String initialPath = "/table_users/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0";
-    String flattenPath = "/table_users/1_000000_0";
+    // The flattened name matches ORIGINAL_PATTERN_COPY ([0-9]+_[0-9]+_copy_[0-9]+) so a
+    // subsequent non-ACID→ACID conversion isn't rejected by the metastore validator.
+    String flattenPath = "/table_users/000000_0_copy_1";
 
     MockFileSystem.MockFile file1 = new MockFileSystem.MockFile("mock://" + initialPath, 0, new byte[1]);
     MockFileSystem fs = new MockFileSystem(new Configuration(), file1);