Delay index scan collect to query execution time

Signed-off-by: Sean Kao <[email protected]>
opensearch-project · Nov 14, 2023 · deb5fe8 · deb5fe8
1 parent 2713f08
commit deb5fe8
Show file tree

Hide file tree

Showing 4 changed files with 11 additions and 13 deletions.
diff --git a/build.sbt b/build.sbt
@@ -141,7 +141,7 @@ lazy val flintSparkIntegration = (project in file("flint-spark-integration"))
     },
     assembly / assemblyExcludedJars := {
       val cp = (assembly / fullClasspath).value
-      cp filter { file => file.data.getName.contains("AWSLogsDALConnectorSpark")}
+      cp filter { file => file.data.getName.contains("LogsConnectorSpark")}
     },
     assembly / test := (Test / test).value)
 

diff --git a/flint-spark-integration/lib/AWSLogsDALConnectorSpark-1.0.jar b/flint-spark-integration/lib/AWSLogsDALConnectorSpark-1.0.jar
diff --git a/flint-spark-integration/lib/LogsConnectorSpark-1.0.jar b/flint-spark-integration/lib/LogsConnectorSpark-1.0.jar
diff --git a/...ion/src/main/scala/org/opensearch/flint/spark/skipping/ApplyFlintSparkSkippingIndex.scala b/...ion/src/main/scala/org/opensearch/flint/spark/skipping/ApplyFlintSparkSkippingIndex.scala
@@ -67,36 +67,34 @@ class ApplyFlintSparkSkippingIndex(flint: FlintSpark) extends Rule[LogicalPlan]
           relation @ DataSourceV2Relation(table, _, Some(catalog), Some(identifier), _))
         if hasNoDisjunction(condition) &&
           // Check if query plan already rewritten
-          table.isInstanceOf[LogsTable] && !table.asInstanceOf[LogsTable].hasLogFileIds() =>
+          table.isInstanceOf[LogsTable] && !table.asInstanceOf[LogsTable].hasFileIndexScan() =>
       val index = flint.describeIndex(getIndexName(catalog, identifier))
       if (index.exists(_.kind == SKIPPING_INDEX_TYPE)) {
         val skippingIndex = index.get.asInstanceOf[FlintSparkSkippingIndex]
         val indexFilter = rewriteToIndexFilter(skippingIndex, condition)
         /*
-         * Replace original LogsTable with a new one with log file idx:
+         * Replace original LogsTable with a new one with file index scan:
          *  Filter(a=b)
          *  |- DataSourceV2Relation(A)
-         *     |- LogsTable <== replaced with a new LogsTable with log file ids
+         *     |- LogsTable <== replaced with a new LogsTable with file index scan
          */
         if (indexFilter.isDefined) {
           val indexScan = flint.queryIndex(skippingIndex.name())
-          val selectedFiles =
+          val selectFileIndexScan =
             // Non hybrid scan
             // TODO: refactor common logic with file-based skipping index
             indexScan
               .filter(new Column(indexFilter.get))
               .select(FILE_PATH_COLUMN)
-              .collect
-              .map(_.getString(0))
 
-          // Construct LogsTable with selectedFiles as its log file ids
-          // It will build scan operator using these log file ids
+          // Construct LogsTable with file index scan
+          // It will build scan operator using log file ids collected from file index scan
           val logsTable = table.asInstanceOf[LogsTable]
           val newTable = new LogsTable(
-            logsTable.getSchema(),
-            logsTable.getOptions(),
-            selectedFiles,
-            logsTable.getProcessedFields())
+            logsTable.schema(),
+            logsTable.options(),
+            selectFileIndexScan,
+            logsTable.processedFields())
           filter.copy(child = relation.copy(table = newTable))
         } else {
           filter