Feature/datafusion (#20134)

* No-op load writer files implementation

* Implemented flush conditions and improved engine failure handling

---------

Co-authored-by: Shashank Gowri <[email protected]>

Author: shank9918

modules/parquet-data-format/src/main/java/com/parquet/parquetdataformat/engine/ParquetExecutionEngine.java

@@ -10,31 +10,24 @@
 import org.apache.arrow.vector.types.pojo.Schema;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
-import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.index.engine.exec.DataFormat;
 import org.opensearch.index.engine.exec.IndexingExecutionEngine;
 import org.opensearch.index.engine.exec.Merger;
 import org.opensearch.index.engine.exec.RefreshInput;
 import org.opensearch.index.engine.exec.RefreshResult;
 import org.opensearch.index.engine.exec.Writer;
-import org.opensearch.index.engine.exec.WriterFileSet;
+import org.opensearch.index.engine.exec.coord.CatalogSnapshot;
 import org.opensearch.index.shard.ShardPath;
 
-import java.io.Closeable;
 import java.io.IOException;
-import java.nio.file.DirectoryStream;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-import java.util.stream.StreamSupport;
 
 import static com.parquet.parquetdataformat.engine.ParquetDataFormat.PARQUET_DATA_FORMAT;
 
@@ -72,11 +65,9 @@ public class ParquetExecutionEngine implements IndexingExecutionEngine<ParquetDa
     private static final Logger logger = LogManager.getLogger(ParquetExecutionEngine.class);
 
     public static final String FILE_NAME_PREFIX = "_parquet_file_generation";
-    private static final Pattern FILE_PATTERN = Pattern.compile(".*_(\\d+)\\.parquet$", Pattern.CASE_INSENSITIVE);
     public static final String FILE_NAME_EXT = ".parquet";
 
     private final Supplier<Schema> schema;
-    private final List<WriterFileSet> filesWrittenAlready = new ArrayList<>();
     private final ShardPath shardPath;
     private final ParquetMerger parquetMerger = new ParquetMergeExecutor(CompactionStrategy.RECORD_BATCH);
     private final ArrowBufferPool arrowBufferPool;
@@ -88,24 +79,12 @@ public ParquetExecutionEngine(Settings settings, Supplier<Schema> schema, ShardP
     }
 
     @Override
-    public void loadWriterFiles() throws IOException {
-        try (DirectoryStream<Path> stream = Files.newDirectoryStream(shardPath.getDataPath(), "*" + FILE_NAME_EXT)) {
-            StreamSupport.stream(stream.spliterator(), false)
-                .map(Path::getFileName)
-                .map(Path::toString)
-                .map(FILE_PATTERN::matcher)
-                .filter(Matcher::matches)
-                .map(m -> WriterFileSet.builder()
-                    .directory(shardPath.getDataPath())
-                    .writerGeneration(Long.parseLong(m.group(1)))
-                    .addFile(m.group(0))
-                    .build())
-                .forEach(filesWrittenAlready::add);
-        }
+    public void loadWriterFiles(CatalogSnapshot catalogSnapshot) {
+        // Noop, as refresh is handled in layers above
     }
 
     @Override
-    public void deleteFiles(Map<String,Collection<String>> filesToDelete) throws IOException {
+    public void deleteFiles(Map<String, Collection<String>> filesToDelete) {
         if (filesToDelete.get(PARQUET_DATA_FORMAT.name()) != null) {
             Collection<String> parquetFilesToDelete = filesToDelete.get(PARQUET_DATA_FORMAT.name());
             for (String fileName : parquetFilesToDelete) {
@@ -127,7 +106,7 @@ public List<String> supportedFieldTypes() {
     }
 
     @Override
-    public Writer<ParquetDocumentInput> createWriter(long writerGeneration) throws IOException {
+    public Writer<ParquetDocumentInput> createWriter(long writerGeneration) {
         String fileName = Path.of(shardPath.getDataPath().toString(), FILE_NAME_PREFIX + "_" + writerGeneration + FILE_NAME_EXT).toString();
         return new ParquetWriter(fileName, schema.get(), writerGeneration, arrowBufferPool);
     }
@@ -138,10 +117,9 @@ public Merger getMerger() {
     }
 
     @Override
-    public RefreshResult refresh(RefreshInput refreshInput) throws IOException {
-        RefreshResult refreshResult = new RefreshResult();
-        // NO-OP, as refresh is being handled at CompositeIndexingExecution Engin
-        return refreshResult;
+    public RefreshResult refresh(RefreshInput refreshInput) {
+        // NO-OP, as refresh is being handled at CompositeIndexingExecutionEngine
+        return new RefreshResult();
     }
 
     @Override

server/src/main/java/org/opensearch/index/engine/Engine.java

@@ -1149,34 +1149,13 @@ public abstract void forceMerge(
      */
     public abstract SafeCommitInfo getSafeCommitInfo();
 
-    /**
-     * If the specified throwable contains a fatal error in the throwable graph, such a fatal error will be thrown. Callers should ensure
-     * that there are no catch statements that would catch an error in the stack as the fatal error here should go uncaught and be handled
-     * by the uncaught exception handler that we install during bootstrap. If the specified throwable does indeed contain a fatal error,
-     * the specified message will attempt to be logged before throwing the fatal error. If the specified throwable does not contain a fatal
-     * error, this method is a no-op.
-     *
-     * @param maybeMessage the message to maybe log
-     * @param maybeFatal   the throwable that maybe contains a fatal error
-     */
-    @SuppressWarnings("finally")
-    private void maybeDie(final String maybeMessage, final Throwable maybeFatal) {
-        ExceptionsHelper.maybeError(maybeFatal).ifPresent(error -> {
-            try {
-                logger.error(maybeMessage, error);
-            } finally {
-                throw error;
-            }
-        });
-    }
-
     /**
      * fail engine due to some error. the engine will also be closed.
      * The underlying store is marked corrupted iff failure is caused by index corruption
      */
     public void failEngine(String reason, @Nullable Exception failure) {
         if (failure != null) {
-            maybeDie(reason, failure);
+            maybeDie(logger, reason, failure);
         }
         if (failEngineLock.tryLock()) {
             try {
@@ -1285,19 +1264,6 @@ protected boolean maybeFailEngine(String source, Exception e) {
         return false;
     }
 
-    /**
-     * Event listener for the engine
-     *
-     * @opensearch.api
-     */
-    @PublicApi(since = "1.0.0")
-    public interface EventListener {
-        /**
-         * Called when a fatal exception occurred
-         */
-        default void onFailedEngine(String reason, @Nullable Exception e) {}
-    }
-
     /**
      * Supplier for the searcher
      *

server/src/main/java/org/opensearch/index/engine/exec/IndexingExecutionEngine.java

@@ -8,6 +8,7 @@
 
 package org.opensearch.index.engine.exec;
 
+import org.opensearch.index.engine.exec.coord.CatalogSnapshot;
 import org.opensearch.index.shard.ShardPath;
 
 import java.io.Closeable;
@@ -29,11 +30,12 @@ Writer<? extends DocumentInput<?>> createWriter(long writerGeneration)
 
     DataFormat getDataFormat();
 
-    void loadWriterFiles() throws IOException;
+    void loadWriterFiles(CatalogSnapshot catalogSnapshot)
+        throws IOException; // Bootstrap hook to make engine aware of previously written files from CatalogSnapshot
 
     default long getNativeBytesUsed() {
         return 0;
     }
 
-    void deleteFiles(Map<String,Collection<String>> filesToDelete) throws IOException;
+    void deleteFiles(Map<String, Collection<String>> filesToDelete) throws IOException;
 }

server/src/main/java/org/opensearch/index/engine/exec/bridge/Indexer.java

@@ -8,7 +8,12 @@
 
 package org.opensearch.index.engine.exec.bridge;
 
+import org.apache.logging.log4j.Logger;
+import org.opensearch.ExceptionsHelper;
+import org.opensearch.common.Nullable;
 import org.opensearch.common.annotation.PublicApi;
+import org.opensearch.common.unit.TimeValue;
+import org.opensearch.core.common.unit.ByteSizeValue;
 import org.opensearch.index.engine.Engine;
 import org.opensearch.index.engine.EngineException;
 import org.opensearch.index.engine.SafeCommitInfo;
@@ -27,10 +32,32 @@
 @PublicApi(since = "1.0.0")
 public interface Indexer {
 
+    /**
+     * Perform document index operation on the engine
+     * @param index operation to perform
+     * @return {@link Engine.IndexResult} containing updated translog location, version and
+     * document specific failures
+     *
+     * Note: engine level failures (i.e. persistent engine failures) are thrown
+     */
     Engine.IndexResult index(Engine.Index index) throws IOException;
 
+    /**
+     * Perform document delete operation on the engine
+     * @param delete operation to perform
+     * @return {@link Engine.DeleteResult} containing updated translog location, version and
+     * document specific failures
+     *
+     * Note: engine level failures (i.e. persistent engine failures) are thrown
+     */
     Engine.DeleteResult delete(Engine.Delete delete) throws IOException;
 
+    /**
+     * Perform a no-op equivalent operation on the engine.
+     * @param noOp
+     * @return
+     * @throws IOException
+     */
     Engine.NoOpResult noOp(Engine.NoOp noOp) throws IOException;
 
     /**
@@ -42,10 +69,22 @@ public interface Indexer {
      */
     int countNumberOfHistoryOperations(String source, long fromSeqNo, long toSeqNumber) throws IOException;
 
+    /**
+     * @param reason why is the history requested
+     * @param startingSeqNo sequence number beyond which history should exist
+     * @return tru iff minimum retained sequence number during indexing is not less than startingSeqNo
+     */
     boolean hasCompleteOperationHistory(String reason, long startingSeqNo);
 
+    /**
+     * Total amount of RAM bytes used for active indexing to buffer unflushed documents.
+     */
     long getIndexBufferRAMBytesUsed();
 
+    /**
+     * @param verbose unused param
+     * @return list of segments the indexer is aware of (previously created + new ones)
+     */
     List<Segment> segments(boolean verbose);
 
     /**
@@ -103,9 +142,22 @@ public interface Indexer {
      */
     void advanceMaxSeqNoOfUpdatesOrDeletes(long maxSeqNoOfUpdatesOnPrimary);
 
+    /**
+     * @return Time in nanos for last write through the indexer, always increasing.
+     */
+    long getLastWriteNanos();
+
+    /**
+     * Fills up the local checkpoints history with no-ops until the local checkpoint
+     * and the max seen sequence ID are identical.
+     * @param primaryTerm the shards primary term this indexer was created for
+     * @return the number of no-ops added
+     */
     int fillSeqNoGaps(long primaryTerm) throws IOException;
 
-    // File format methods follow below
+    /**
+     * Performs a force merge operation on this engine.
+     */
     void forceMerge(
         boolean flush,
         int maxNumSegments,
@@ -115,12 +167,35 @@ void forceMerge(
         String forceMergeUUID
     ) throws EngineException, IOException;
 
+    /**
+     * Applies changes to input settings.
+     */
+    void onSettingsChanged(TimeValue translogRetentionAge, ByteSizeValue translogRetentionSize, long softDeletesRetentionOps);
+
+    /**
+     * Flushes active indexing buffer to disk.
+     */
     void writeIndexingBuffer() throws EngineException;
 
+    /**
+     * Creates segments for data in buffers, and make them available for search.
+     */
     void refresh(String source) throws EngineException;
 
+    /**
+     * Commits the data and state to disk, resulting in documents being persisted onto the underlying formats.
+     */
     void flush(boolean force, boolean waitIfOngoing) throws EngineException;
 
+    /**
+     * Checks if data should be committed to disk, mainly based on translog thresholds.
+     * @return true iff flush should trigger.
+     */
+    boolean shouldPeriodicallyFlush();
+
+    /**
+     * Returns info about the safe commit.
+     */
     SafeCommitInfo getSafeCommitInfo();
 
     // Translog methods follow below
@@ -135,6 +210,42 @@ Translog.Snapshot newChangesSnapshot(String source, long fromSeqNo, long toSeqNo
 
     void flushAndClose() throws IOException;
 
+    void failEngine(String reason, @Nullable Exception failure);
+
+    /**
+     * If the specified throwable contains a fatal error in the throwable graph, such a fatal error will be thrown. Callers should ensure
+     * that there are no catch statements that would catch an error in the stack as the fatal error here should go uncaught and be handled
+     * by the uncaught exception handler that we install during bootstrap. If the specified throwable does indeed contain a fatal error,
+     * the specified message will attempt to be logged before throwing the fatal error. If the specified throwable does not contain a fatal
+     * error, this method is a no-op.
+     *
+     * @param maybeMessage the message to maybe log
+     * @param maybeFatal   the throwable that maybe contains a fatal error
+     */
+    @SuppressWarnings("finally")
+    default void maybeDie(final Logger logger, final String maybeMessage, final Throwable maybeFatal) {
+        ExceptionsHelper.maybeError(maybeFatal).ifPresent(error -> {
+            try {
+                logger.error(maybeMessage, error);
+            } finally {
+                throw error;
+            }
+        });
+    }
+
+    /**
+     * Event listener for the engine
+     *
+     * @opensearch.api
+     */
+    @PublicApi(since = "1.0.0")
+    public interface EventListener {
+        /**
+         * Called when a fatal exception occurred
+         */
+        default void onFailedEngine(String reason, @Nullable Exception e) {}
+    }
+
     /**
      * Reads the current stored history ID from commit data.
      */

server/src/main/java/org/opensearch/index/engine/exec/commit/Committer.java

@@ -14,7 +14,6 @@
 import java.io.Closeable;
 import java.io.IOException;
 import java.util.Map;
-import java.util.Optional;
 
 public interface Committer extends Closeable {
 

server/src/main/java/org/opensearch/index/engine/exec/commit/LuceneCommitEngine.java

@@ -26,11 +26,8 @@
 import java.nio.file.Path;
 import java.util.Collection;
 import java.util.Map;
-import java.util.Optional;
 import java.util.function.LongSupplier;
 
-import static org.opensearch.index.engine.exec.coord.CatalogSnapshot.CATALOG_SNAPSHOT_KEY;
-
 public class LuceneCommitEngine implements Committer {
 
     private final Logger logger;