Fix bug when SWA hdfs and local paths without data.avro.json extensio… (#1130)

rakeshkashyap123 · Rakesh Kashyap Hanasoge Padmanabha · web-flow · commit 65702dc1032f · 2023-04-05T11:23:43.000-07:00
* fix bug when SWA hdfs and local paths without data.avro.json extensions are included for evaluation

* try

* Fix tests

* revert test file

* Add tests

* Add private classifier to variable

* fix test

* fix test

---------

Co-authored-by: Rakesh Kashyap Hanasoge Padmanabha &lt;rkashyap@rkashyap-mn3.linkedin.biz&gt;
diff --git a/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/HdfsPathChecker.scala b/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/HdfsPathChecker.scala
@@ -9,4 +9,13 @@ private[offline] class HdfsPathChecker extends PathChecker {
   override def isMock(path: String): Boolean = false
 
   override def exists(path: String): Boolean = HdfsUtils.exists(path)
+
+  /**
+   * Check whether the given path has any file content. If it is directory, then ensure it has atleast one file, else check
+   * the file size is not empty.
+   *
+   * @param path input path
+   * @return true if the path is non empty.
+   */
+  override def nonEmpty(path: String): Boolean = HdfsUtils.nonEmpty(path)
 }
diff --git a/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/LocalPathChecker.scala b/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/LocalPathChecker.scala
@@ -4,12 +4,13 @@ import com.linkedin.feathr.offline.util.{HdfsUtils, LocalFeatureJoinUtils, Sourc
 import com.linkedin.feathr.offline.source.dataloader.DataLoaderHandler
 import org.apache.hadoop.conf.Configuration
 
+import java.io.File
+
 /**
  * path checker for local test files.
  * @param hadoopConf hadoop configuration
  */
 private[offline] class LocalPathChecker(hadoopConf: Configuration, dataLoaderHandlers: List[DataLoaderHandler]) extends PathChecker {
-
   private val TEST_AVRO_JSON_FILE = "/data.avro.json"
 
   /**
@@ -35,7 +36,7 @@ private[offline] class LocalPathChecker(hadoopConf: Configuration, dataLoaderHan
         if (dataLoaderHandler.validatePath(path)) {
           isExternalDataSourceFlag = true
           break
-        } 
+        }
       }
     }
     isExternalDataSourceFlag
@@ -53,4 +54,25 @@ private[offline] class LocalPathChecker(hadoopConf: Configuration, dataLoaderHan
     if (getClass.getClassLoader.getResource(path + TEST_AVRO_JSON_FILE) != null) return true
     false
   }
+
+  /**
+   * Checks if the given path is non empty. If the path is a directory, check if it has files listed under it. Else, just check the
+   * file length.
+   * @param path input path
+   * @return true if the path is non empty.
+   */
+  override def nonEmpty(path: String): Boolean = {
+    if (!isExternalDataSource(path) && HdfsUtils.nonEmpty(path)) return true
+    val filePath = if (LocalFeatureJoinUtils.getMockPathIfExist(path, hadoopConf, None).isDefined) {
+      new File(LocalFeatureJoinUtils.getMockPathIfExist(path, hadoopConf, None).get)
+    } else {
+      new File(getClass.getClassLoader.getResource(path).toURI)
+    }
+
+    if (filePath.isDirectory) { // we do not need to check if the file have any length
+      filePath.listFiles().length > 0
+    } else {
+      filePath.length() > 0
+    }
+  }
 }
diff --git a/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/PathChecker.scala b/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/PathChecker.scala
@@ -20,6 +20,14 @@ private[offline] trait PathChecker {
    * @return true if the path exists.
    */
   def exists(path: String) : Boolean
+
+  /**
+   * Check whether the given path has any file content. If it is directory, then ensure it has atleast one file, else check
+   * the file size is not empty.
+   * @param path input path
+   * @return true if the path is non empty.
+   */
+  def nonEmpty(path: String) : Boolean
 }
 
 /**
diff --git a/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/TimeBasedHdfsPathGenerator.scala b/feathr-impl/src/main/scala/com/linkedin/feathr/offline/source/pathutil/TimeBasedHdfsPathGenerator.scala
@@ -9,7 +9,6 @@ import java.time.format.DateTimeFormatter
  * @param pathChecker the path checker is used to check whether a file path exists.
  */
 private[offline] class TimeBasedHdfsPathGenerator(pathChecker: PathChecker) {
-
   /**
    * Helper function for generating file names for daily and hourly format data
    * Supported path format include:
@@ -35,7 +34,7 @@ private[offline] class TimeBasedHdfsPathGenerator(pathChecker: PathChecker) {
         .map(offset => pathInfo.basePath + formatter.format(factDataStartTime.plus(offset, chronUnit)) + postfixPath).distinct
 
     if (ignoreMissingFiles) {
-      filePaths.filter(pathChecker.exists)
+      filePaths.filter(filePath => pathChecker.exists(filePath) && pathChecker.nonEmpty(filePath))
     } else {
       filePaths
     }
diff --git a/feathr-impl/src/main/scala/com/linkedin/feathr/offline/util/HdfsUtils.scala b/feathr-impl/src/main/scala/com/linkedin/feathr/offline/util/HdfsUtils.scala
@@ -1,14 +1,14 @@
 package com.linkedin.feathr.offline.util
 
+import com.linkedin.feathr.offline.util.HdfsUtils.conf
 import org.apache.hadoop.conf.Configuration
-import org.apache.hadoop.fs.{Path,LocatedFileStatus,FileSystem,PathFilter,RemoteIterator}
+import org.apache.hadoop.fs.{FileSystem, LocatedFileStatus, Path, PathFilter, RemoteIterator}
 import org.apache.log4j.{Logger, PatternLayout, WriterAppender}
 
 import java.io.{FileSystem => _, _}
 import java.time.format.DateTimeFormatter
 import java.time.temporal.ChronoUnit
 import java.time.{LocalDateTime, ZoneId, ZoneOffset}
-
 import scala.annotation.tailrec
 
 
@@ -479,6 +479,25 @@ object HdfsUtils {
     }
   }
 
+  /**
+   * For a given input path, check if it is non-empty. If the path is a directory, check if it has any files within the directory.
+   * Otherwise, check if the file is non-empty.
+   *
+   * @param conf      Hadoop Configuration
+   * @param inputPath input path
+   * @return true if nonEmpty
+   */
+  def nonEmpty(inputPath: String, conf: Configuration = conf): Boolean = {
+    val fs = FileSystem.get(conf)
+    val path = new Path(inputPath)
+    if (!exists(inputPath)) return false
+    if (fs.getFileStatus(path).isDirectory) {
+      fs.listStatus(path).length > 0
+    } else {
+      fs.getFileStatus(path).getLen > 0
+    }
+  }
+
   /**
    * Creaete directories as needed
    */
diff --git a/feathr-impl/src/test/scala/com/linkedin/feathr/offline/source/pathutil/TestPathChecker.scala b/feathr-impl/src/test/scala/com/linkedin/feathr/offline/source/pathutil/TestPathChecker.scala
@@ -34,6 +34,9 @@ class TestPathChecker extends TestFeathr with MockitoSugar {
     assertEquals(hdfsPathChecker.isMock("anyPath"), false)
     assertEquals(hdfsPathChecker.exists("src/test/resources/anchor1-source.csv"), true)
     assertEquals(hdfsPathChecker.exists("non_existing_path"), false)
+    assertEquals(hdfsPathChecker.nonEmpty("src/test/resources/generation/daily/2019/05/19"), true)
+    assertEquals(hdfsPathChecker.nonEmpty("src/test/resources/generation/daily/2019/05/18"), false)
+    assertEquals(hdfsPathChecker.nonEmpty("src/test/resources/anchor1-source.csv"), true)
   }
 
 
@@ -44,6 +47,8 @@ class TestPathChecker extends TestFeathr with MockitoSugar {
     assertEquals(localPathChecker.exists("anchor1-source.csv"), true)
     assertEquals(localPathChecker.exists("generation/daily/2019/05/19"), true)
     assertEquals(localPathChecker.exists("non-existing_path"), false)
+    assertEquals(localPathChecker.nonEmpty("generation/daily/2019/05/19"), true)
+    assertEquals(localPathChecker.nonEmpty("anchor1-source.csv"), true)
   }
 
   @Test(description = "test isExternalDataSource method for LocalPathChecker")
diff --git a/feathr-impl/src/test/scala/com/linkedin/feathr/offline/source/pathutil/TestTimeBasedHdfsPathGenerator.scala b/feathr-impl/src/test/scala/com/linkedin/feathr/offline/source/pathutil/TestTimeBasedHdfsPathGenerator.scala
@@ -57,11 +57,33 @@ class TestTimeBasedHdfsPathGenerator extends TestFeathr with MockitoSugar{
     val pathInfo = PathInfo("src/test/resources/generation/daily/", DateTimeResolution.DAILY, "yyyy/MM/dd")
     when(mockPathChecker.exists("src/test/resources/generation/daily/2019/05/18")).thenReturn(false)
     when(mockPathChecker.exists("src/test/resources/generation/daily/2019/05/19")).thenReturn(true)
+    when(mockPathChecker.nonEmpty("src/test/resources/generation/daily/2019/05/19")).thenReturn(true)
     val interval = TestUtils.createDailyInterval("2019-05-18", "2019-05-20")
     val pathList = pathGenerator.generate(pathInfo, interval, true)
     assertEquals(pathList.toList, List("src/test/resources/generation/daily/2019/05/19"))
     verify(mockPathChecker).exists("src/test/resources/generation/daily/2019/05/18")
     verify(mockPathChecker).exists("src/test/resources/generation/daily/2019/05/19")
+    verify(mockPathChecker).nonEmpty("src/test/resources/generation/daily/2019/05/19")
+    verifyNoMoreInteractions(mockPathChecker)
+  }
+
+  @Test(description = "test ignore missing files and empty folders")
+  def testGenerateDailyFilesWithEmptyFolders(): Unit = {
+    val mockPathChecker = mock[PathChecker]
+    val pathGenerator = new TimeBasedHdfsPathGenerator(mockPathChecker)
+
+    val pathInfo = PathInfo("src/test/resources/generation/daily/", DateTimeResolution.DAILY, "yyyy/MM/dd")
+    when(mockPathChecker.exists("src/test/resources/generation/daily/2019/05/18")).thenReturn(true)
+    when(mockPathChecker.exists("src/test/resources/generation/daily/2019/05/19")).thenReturn(true)
+    when(mockPathChecker.nonEmpty("src/test/resources/generation/daily/2019/05/18")).thenReturn(false)
+    when(mockPathChecker.nonEmpty("src/test/resources/generation/daily/2019/05/19")).thenReturn(true)
+    val interval = TestUtils.createDailyInterval("2019-05-18", "2019-05-20")
+    val pathList = pathGenerator.generate(pathInfo, interval, true)
+    assertEquals(pathList.toList, List("src/test/resources/generation/daily/2019/05/19"))
+    verify(mockPathChecker).exists("src/test/resources/generation/daily/2019/05/18")
+    verify(mockPathChecker).exists("src/test/resources/generation/daily/2019/05/19")
+    verify(mockPathChecker).nonEmpty("src/test/resources/generation/daily/2019/05/19")
+    verify(mockPathChecker).nonEmpty("src/test/resources/generation/daily/2019/05/18")
     verifyNoMoreInteractions(mockPathChecker)
   }
 
